1 Commits

Author SHA1 Message Date
985464dcf9 Fail loudly on set-but-unparseable env config values (closes #80)
All checks were successful
check / check (push) Successful in 6m3s
The config env helpers silently substituted the documented default
whenever a variable was set but could not be parsed, so a typo in an
operator-supplied value produced a running daemon with configuration
nobody asked for instead of a startup failure. `PORT=eighty` quietly
listened on 8080 and `DEBUG=ture` quietly disabled debug logging.

Defaults now apply only to variables that are unset or empty. Any
variable that is set but unparseable is a hard error that names the
key and the offending value and aborts startup through fx.

- add `envPositiveInt` with `ErrNonPositiveValue`, copied verbatim
  from the definition on the unmerged #87 so that rebasing after it
  lands is a delete-one-copy operation rather than a semantic merge
- remove `envInt` entirely; `PORT` is parsed by a new `envPort`,
  which adds the TCP upper bound (`ErrInvalidPort`, 1..65535)
- change `envBool` to return an error and parse with
  `strconv.ParseBool`, so `yes`, `on`, and typos are rejected rather
  than silently treated as false; callers are `DEBUG` and
  `MAINTENANCE_MODE`
- move env loading into `loadFromEnv`, with the environment check
  extracted to `resolveEnvironment` (also as #87 defines it), keeping
  `New` within the funlen budget

`envString` parses nothing and `envDuration` was already fail-loud,
so both are unchanged. A repo-wide audit of `os.Getenv`/`os.LookupEnv`
found no parse sites outside `internal/config`.

Tests cover each helper with a table (unset, valid, set-but-invalid)
plus `config.New`-level cases proving a bad `PORT`, `DEBUG`, or
`MAINTENANCE_MODE` aborts startup while unset variables still get
their defaults. README documents the fail-loud rule and the accepted
boolean spellings.
2026-08-09 01:45:27 +00:00
19 changed files with 611 additions and 2602 deletions

View File

@@ -89,10 +89,27 @@ TTY detection, and security headers are always applied.
| `PORT` | HTTP listen port | `8080` | | `PORT` | HTTP listen port | `8080` |
| `DATA_DIR` | Directory for all SQLite databases | `/var/lib/webhooker` | | `DATA_DIR` | Directory for all SQLite databases | `/var/lib/webhooker` |
| `DEBUG` | Enable debug logging | `false` | | `DEBUG` | Enable debug logging | `false` |
| `MAINTENANCE_MODE` | Serve the maintenance page | `false` |
| `METRICS_USERNAME` | Basic auth username for `/metrics` | `""` | | `METRICS_USERNAME` | Basic auth username for `/metrics` | `""` |
| `METRICS_PASSWORD` | Basic auth password for `/metrics` | `""` | | `METRICS_PASSWORD` | Basic auth password for `/metrics` | `""` |
| `SENTRY_DSN` | Sentry error reporting DSN | `""` | | `SENTRY_DSN` | Sentry error reporting DSN | `""` |
#### Invalid values abort startup
The defaults above apply **only** to variables that are unset (or set
to an empty string). A variable that is set but cannot be parsed is a
fatal configuration error: webhooker logs the offending variable and
its value and refuses to start, rather than silently running with a
substituted default. `PORT=eighty`, `DEBUG=ture`, and
`RETENTION_SWEEP_INTERVAL=1 hour` all abort startup. `PORT` must
additionally be a number in the range 165535.
Boolean variables (`DEBUG`, `MAINTENANCE_MODE`) accept exactly the
spellings Go's `strconv.ParseBool` accepts — `1`, `t`, `T`, `TRUE`,
`true`, `True`, `0`, `f`, `F`, `FALSE`, `false`, `False` — and nothing
else. `yes`, `on`, and `off` are rejected rather than quietly treated
as false.
On first startup, webhooker automatically generates a cryptographically On first startup, webhooker automatically generates a cryptographically
secure session encryption key and stores it in the database. This key secure session encryption key and stores it in the database. This key
persists across restarts — no manual key management is needed. persists across restarts — no manual key management is needed.
@@ -531,36 +548,6 @@ older than the expiry are pruned each time the archive is (re)opened. An
archive write failure is never silent success: the delivery records a archive write failure is never silent success: the delivery records a
failed attempt with the error and is marked failed. failed attempt with the error and is marked failed.
Because reopens only happen on writes, an archive belonging to a webhook
that has stopped receiving events would never be pruned. A background
**archive sweeper** closes that gap: on the same interval as the event
retention reaper (`RETENTION_SWEEP_INTERVAL`) it prunes every archive
whose database target declares a positive expiry, whether or not the
webhook is still receiving traffic. The sweep never creates an archive —
a webhook whose archive file does not yet exist is skipped, not
initialised — it takes the same per-webhook lock the write path uses, so
it can never interleave with a write, and it leaves the archive closed
afterwards so the move-the-file-away workflow keeps working. Archives
with no expiry, or the expiry `never`, are not touched by the sweep at
all.
Note that a webhook has one archive file but may carry more than one
`database` target, each with its own `expiry`. The shortest expiry
configured on any of them therefore governs the whole archive, and the
sweep applies it whether or not the webhook is still receiving events.
Configure a single `database` target per webhook unless you intend that.
Deleting a webhook releases its archive: the delivery engine's cached
archive writer is dropped and its file handle closed, so nothing lingers
after the webhook is gone. The archive **file itself is deliberately
left on disk**. Unlike the event database — per-webhook working storage
that is hard-deleted with the webhook — an archive is long-term storage
an operator may still want to keep or move away for offline retention,
and destroying it as a side effect of deleting a webhook would be
unrecoverable. Removing `archive-{webhookID}.db` is the operator's call.
Deleting a webhook's last `database` target releases the writer the same
way, and for the same reason leaves the file alone.
The **Slack target type** sends webhook events as formatted messages to The **Slack target type** sends webhook events as formatted messages to
any Slack-compatible incoming webhook URL (works with Slack, Mattermost, any Slack-compatible incoming webhook URL (works with Slack, Mattermost,
and other compatible services). Each message includes event metadata and other compatible services). Each message includes event metadata

32
TODO.md
View File

@@ -12,28 +12,26 @@
pre-1.0. No git tags exist. main (4f5ecb1) is a working webhook proxy pre-1.0. No git tags exist. main (4f5ecb1) is a working webhook proxy
with auth, CSRF/SSRF protections, login rate limiting, Slack target, with auth, CSRF/SSRF protections, login rate limiting, Slack target,
event retention (#63), the database archiving target (#43), the admin policy compliance (#6), pinned lint tooling (#55), a per-webhook event
password change flow (#65), policy compliance (#6), and pinned lint retention reaper (#63), and fail-loud configuration parsing (#80). Note:
tooling (#55). Note: TODO.md was deliberately deleted from this repo in TODO.md was deliberately deleted from this repo in f9a9569 (2026-03-01,
f9a9569 (2026-03-01, #6); its content was folded into the README TODO #6); its content was folded into the README TODO section, which this
section, which this draft reconstructs as of 2026-07-06. draft reconstructs as of 2026-07-06.
# Next Step # Next Step
Implement automatic event retention cleanup based on retention_days: a Manual event redelivery from the web UI (replay is a core promised
periodic maintenance job that deletes Events, Deliveries, and capability in the README rationale).
DeliveryResults older than the parent webhook's retention_days from each
per-webhook event database. The field exists on the Webhook model and
the README promises the behavior, but nothing enforces it, so event
databases currently grow without bound.
# Completed Steps # Completed Steps
- 2026-08-09 Archive writer lifecycle (#89): deleting a webhook (or its - 2026-08-09 Configuration parsing fails loudly on set-but-unparseable
last `database` target) evicts the cached archive writer and closes environment values: `envInt` removed in favour of `envPositiveInt`
its handle while deliberately leaving `archive-{webhookID}.db` on plus a `PORT` range check, `envBool` now parses with
disk, and a new `ArchiveSweeper` prunes idle archives on the existing `strconv.ParseBool`, and defaults apply only to unset variables (#80)
`RETENTION_SWEEP_INTERVAL` without ever creating an archive file - 2026-08-07 Automatic event retention cleanup based on
`retention_days`, deleting expired events, deliveries, and delivery
results from each per-webhook event database (#63)
- 2026-08-07 Update golangci-lint to v2.12.2 (Docker image digest in - 2026-08-07 Update golangci-lint to v2.12.2 (Docker image digest in
`Dockerfile`, release-archive sha256 pins in `script/bootstrap`), `Dockerfile`, release-archive sha256 pins in `script/bootstrap`),
adopt the canonical `.golangci.yml` (v2 `linters.settings` layout so adopt the canonical `.golangci.yml` (v2 `linters.settings` layout so
@@ -62,8 +60,6 @@ databases currently grow without bound.
# Future Steps # Future Steps
- Manual event redelivery from the web UI (replay is a core promised
capability in the README rationale)
- Delivery status and retry management UI - Delivery status and retry management UI
- Per-webhook rate limiting in the receiver handler (per-webhook config - Per-webhook rate limiting in the receiver handler (per-webhook config
plus handler enforcement; global limits must not apply to receiver plus handler enforcement; global limits must not apply to receiver

View File

@@ -40,15 +40,9 @@ func main() {
handlers.New, handlers.New,
middleware.New, middleware.New,
delivery.New, delivery.New,
delivery.NewArchiveSweeper,
// Wire *delivery.Engine as delivery.Notifier so the // Wire *delivery.Engine as delivery.Notifier so the
// webhook handler can notify the engine of new deliveries. // webhook handler can notify the engine of new deliveries.
func(e *delivery.Engine) delivery.Notifier { return e }, func(e *delivery.Engine) delivery.Notifier { return e },
// Wire *delivery.Engine as delivery.WebhookEvictor so
// deleting a webhook releases its archive writer.
func(e *delivery.Engine) delivery.WebhookEvictor {
return e
},
server.New, server.New,
), ),
fx.Invoke( fx.Invoke(
@@ -56,7 +50,6 @@ func main() {
*server.Server, *server.Server,
*delivery.Engine, *delivery.Engine,
*database.RetentionReaper, *database.RetentionReaper,
*delivery.ArchiveSweeper,
) { ) {
}, },
), ),

View File

@@ -7,7 +7,6 @@ import (
"log/slog" "log/slog"
"os" "os"
"strconv" "strconv"
"strings"
"time" "time"
"go.uber.org/fx" "go.uber.org/fx"
@@ -31,12 +30,24 @@ const (
// defaultRetentionSweepInterval is how often the retention // defaultRetentionSweepInterval is how often the retention
// reaper deletes events older than each webhook's RetentionDays. // reaper deletes events older than each webhook's RetentionDays.
defaultRetentionSweepInterval = time.Hour defaultRetentionSweepInterval = time.Hour
// maxPort is the highest valid TCP port number. The lower
// bound (at least 1) is enforced by envPositiveInt.
maxPort = 65535
) )
// ErrInvalidEnvironment is returned when WEBHOOKER_ENVIRONMENT // ErrInvalidEnvironment is returned when WEBHOOKER_ENVIRONMENT
// contains an unrecognised value. // contains an unrecognised value.
var ErrInvalidEnvironment = errors.New("invalid environment") var ErrInvalidEnvironment = errors.New("invalid environment")
// ErrNonPositiveValue is returned when an environment variable that
// requires a positive integer is set to zero or a negative number.
var ErrNonPositiveValue = errors.New("value must be positive")
// ErrInvalidPort is returned when an environment variable holding a
// TCP port number is set above the valid port range.
var ErrInvalidPort = errors.New("invalid port")
//nolint:revive // ConfigParams is a standard fx naming convention. //nolint:revive // ConfigParams is a standard fx naming convention.
type ConfigParams struct { type ConfigParams struct {
fx.In fx.In
@@ -81,27 +92,81 @@ func envString(key string) string {
} }
// envBool returns the value of the named environment variable // envBool returns the value of the named environment variable
// parsed as a boolean. Returns defaultValue if not set. // parsed as a boolean. Returns defaultValue if not set. If the
func envBool(key string, defaultValue bool) bool { // variable is set but cannot be parsed, it returns a wrapped error
if v := os.Getenv(key); v != "" { // naming the key and the bad value, so startup fails loudly rather
return strings.EqualFold(v, "true") || v == "1" // than silently falling back to the default.
//
// Parsing is strconv.ParseBool, which accepts 1, t, T, TRUE, true,
// True, 0, f, F, FALSE, false and False. Anything else — "yes",
// "on", or a typo like "ture" — is an error rather than a silent
// false.
func envBool(key string, defaultValue bool) (bool, error) {
v := os.Getenv(key)
if v == "" {
return defaultValue, nil
} }
return defaultValue b, err := strconv.ParseBool(v)
if err != nil {
return false, fmt.Errorf(
"invalid boolean for %s: %q: %w", key, v, err,
)
}
return b, nil
}
// envPositiveInt returns the value of the named environment variable
// parsed as a positive integer. Returns defaultValue if not set. If
// the variable is set but cannot be parsed, or parses to less than
// one, it returns a wrapped error naming the key and the bad value,
// so startup fails loudly rather than silently falling back to the
// default.
func envPositiveInt(
key string,
defaultValue int,
) (int, error) {
v := os.Getenv(key)
if v == "" {
return defaultValue, nil
} }
// envInt returns the value of the named environment variable
// parsed as an integer. Returns defaultValue if not set or
// unparseable.
func envInt(key string, defaultValue int) int {
if v := os.Getenv(key); v != "" {
i, err := strconv.Atoi(v) i, err := strconv.Atoi(v)
if err == nil { if err != nil {
return i return 0, fmt.Errorf(
} "invalid integer for %s: %q: %w", key, v, err,
)
} }
return defaultValue if i < 1 {
return 0, fmt.Errorf(
"%w: %s must be at least 1, got %q",
ErrNonPositiveValue, key, v,
)
}
return i, nil
}
// envPort returns the value of the named environment variable parsed
// as a TCP port number. Returns defaultValue if not set. A set value
// that is unparseable, below 1, or above maxPort is a hard error
// naming the key and the bad value.
func envPort(key string, defaultValue int) (int, error) {
port, err := envPositiveInt(key, defaultValue)
if err != nil {
return 0, err
}
if port > maxPort {
return 0, fmt.Errorf(
"%w: %s must be at most %d, got %d",
ErrInvalidPort, key, maxPort, port,
)
}
return port, nil
} }
// envDuration returns the value of the named environment variable // envDuration returns the value of the named environment variable
@@ -128,32 +193,52 @@ func envDuration(
return d, nil return d, nil
} }
// New creates a Config by reading environment variables. // resolveEnvironment reads WEBHOOKER_ENVIRONMENT, defaulting to
// // dev, and rejects unrecognised values.
//nolint:revive // lc parameter is required by fx even if unused. func resolveEnvironment() (string, error) {
func New(lc fx.Lifecycle, params ConfigParams) (*Config, error) {
log := params.Logger.Get()
// Determine environment from WEBHOOKER_ENVIRONMENT env var,
// default to dev
environment := os.Getenv("WEBHOOKER_ENVIRONMENT") environment := os.Getenv("WEBHOOKER_ENVIRONMENT")
if environment == "" { if environment == "" {
environment = EnvironmentDev environment = EnvironmentDev
} }
// Validate environment
if environment != EnvironmentDev && if environment != EnvironmentDev &&
environment != EnvironmentProd { environment != EnvironmentProd {
return nil, fmt.Errorf( return "", fmt.Errorf(
"%w: WEBHOOKER_ENVIRONMENT must be '%s' or '%s', got '%s'", "%w: WEBHOOKER_ENVIRONMENT must be '%s' or '%s', got '%s'",
ErrInvalidEnvironment, ErrInvalidEnvironment,
EnvironmentDev, EnvironmentProd, environment, EnvironmentDev, EnvironmentProd, environment,
) )
} }
// Parse the retention sweep interval; a set-but-unparseable value return environment, nil
// is a hard error so fx aborts startup rather than silently using }
// the default.
// loadFromEnv builds a Config from the environment. Every value that
// needs parsing fails loudly when it is set but unparseable: the
// documented defaults apply only to variables that are unset (or
// empty), never as a substitute for a value the operator actually
// provided.
func loadFromEnv() (*Config, error) {
environment, err := resolveEnvironment()
if err != nil {
return nil, err
}
port, err := envPort("PORT", defaultPort)
if err != nil {
return nil, err
}
debug, err := envBool("DEBUG", false)
if err != nil {
return nil, err
}
maintenanceMode, err := envBool("MAINTENANCE_MODE", false)
if err != nil {
return nil, err
}
retentionSweepInterval, err := envDuration( retentionSweepInterval, err := envDuration(
"RETENTION_SWEEP_INTERVAL", "RETENTION_SWEEP_INTERVAL",
defaultRetentionSweepInterval, defaultRetentionSweepInterval,
@@ -162,21 +247,36 @@ func New(lc fx.Lifecycle, params ConfigParams) (*Config, error) {
return nil, err return nil, err
} }
// Load configuration values from environment variables return &Config{
s := &Config{
DataDir: envString("DATA_DIR"), DataDir: envString("DATA_DIR"),
Debug: envBool("DEBUG", false), Debug: debug,
MaintenanceMode: envBool("MAINTENANCE_MODE", false), MaintenanceMode: maintenanceMode,
Environment: environment, Environment: environment,
MetricsUsername: envString("METRICS_USERNAME"), MetricsUsername: envString("METRICS_USERNAME"),
MetricsPassword: envString("METRICS_PASSWORD"), MetricsPassword: envString("METRICS_PASSWORD"),
Port: envInt("PORT", defaultPort), Port: port,
SentryDSN: envString("SENTRY_DSN"), SentryDSN: envString("SENTRY_DSN"),
RetentionSweepInterval: retentionSweepInterval, RetentionSweepInterval: retentionSweepInterval,
log: log, }, nil
params: &params,
} }
// New creates a Config by reading environment variables.
//
//nolint:revive // lc parameter is required by fx even if unused.
func New(lc fx.Lifecycle, params ConfigParams) (*Config, error) {
log := params.Logger.Get()
// A set-but-unparseable value anywhere in the environment is a
// hard error, so fx aborts startup rather than running with a
// silently substituted default.
s, err := loadFromEnv()
if err != nil {
return nil, err
}
s.log = log
s.params = &params
// Set default DataDir. All SQLite databases (main application // Set default DataDir. All SQLite databases (main application
// DB and per-webhook event DBs) live here. The same default is // DB and per-webhook event DBs) live here. The same default is
// used regardless of environment; override with DATA_DIR if // used regardless of environment; override with DATA_DIR if

409
internal/config/env_test.go Normal file
View File

@@ -0,0 +1,409 @@
package config_test
import (
"os"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"go.uber.org/fx"
"sneak.berlin/go/webhooker/internal/config"
"sneak.berlin/go/webhooker/internal/globals"
"sneak.berlin/go/webhooker/internal/logger"
)
// testEnvKey is a throwaway variable name used only by the helper
// tables below, so they cannot disturb real configuration.
const testEnvKey = "WEBHOOKER_TEST_VALUE"
// Real configuration variables exercised by the config.New tests.
const (
envKeyPort = "PORT"
envKeyDebug = "DEBUG"
envKeyMaintenanceMode = "MAINTENANCE_MODE"
)
// envBoolCase is one row of the envBool table.
type envBoolCase struct {
name string
set bool
value string
defaultValue bool
expectError bool
expected bool
}
// envBoolCases is the envBool table, kept out of the test body so
// the test itself stays readable.
func envBoolCases() []envBoolCase {
return []envBoolCase{
{
name: "unset uses default false",
defaultValue: false,
expected: false,
},
{
name: "unset uses default true",
defaultValue: true,
expected: true,
},
{
name: "empty uses default true",
set: true,
value: "",
defaultValue: true,
expected: true,
},
{
name: "true is parsed",
set: true,
value: "true",
expected: true,
},
{
name: "one is parsed",
set: true,
value: "1",
expected: true,
},
{
name: "False is parsed",
set: true,
value: "False",
defaultValue: true,
expected: false,
},
{
name: "zero is parsed",
set: true,
value: "0",
defaultValue: true,
expected: false,
},
{
name: "yes is rejected",
set: true,
value: "yes",
expectError: true,
},
{
name: "on is rejected",
set: true,
value: "on",
expectError: true,
},
{
name: "typo is rejected",
set: true,
value: "ture",
expectError: true,
},
}
}
func TestEnvBool(t *testing.T) {
for _, tt := range envBoolCases() {
t.Run(tt.name, func(t *testing.T) {
// Cannot use t.Parallel() here because t.Setenv
// is incompatible with parallel subtests.
if tt.set {
t.Setenv(testEnvKey, tt.value)
} else {
require.NoError(t, os.Unsetenv(testEnvKey))
}
got, err := config.EnvBoolForTest(
testEnvKey, tt.defaultValue,
)
if tt.expectError {
require.Error(t, err)
assert.Contains(t, err.Error(), testEnvKey)
assert.Contains(t, err.Error(), tt.value)
return
}
require.NoError(t, err)
assert.Equal(t, tt.expected, got)
})
}
}
func TestEnvPositiveInt(t *testing.T) {
const defaultValue = 7
tests := []struct {
name string
set bool
value string
expectError bool
errIs error
expected int
}{
{
name: "unset returns the default integer",
expected: defaultValue,
},
{
name: "empty returns the default integer",
set: true,
value: "",
expected: defaultValue,
},
{
name: "positive value is parsed",
set: true,
value: "42",
expected: 42,
},
{
name: "unparseable value is rejected",
set: true,
value: "not-a-number",
expectError: true,
},
{
name: "zero is rejected",
set: true,
value: "0",
expectError: true,
errIs: config.ErrNonPositiveValue,
},
{
name: "negative is rejected",
set: true,
value: "-5",
expectError: true,
errIs: config.ErrNonPositiveValue,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
// Cannot use t.Parallel() here because t.Setenv
// is incompatible with parallel subtests.
if tt.set {
t.Setenv(testEnvKey, tt.value)
} else {
require.NoError(t, os.Unsetenv(testEnvKey))
}
got, err := config.EnvPositiveIntForTest(
testEnvKey, defaultValue,
)
if tt.expectError {
require.Error(t, err)
assert.Contains(t, err.Error(), testEnvKey)
assert.Contains(t, err.Error(), tt.value)
if tt.errIs != nil {
require.ErrorIs(t, err, tt.errIs)
}
return
}
require.NoError(t, err)
assert.Equal(t, tt.expected, got)
})
}
}
func TestEnvPort(t *testing.T) {
const defaultValue = 8080
tests := []struct {
name string
set bool
value string
expectError bool
errIs error
expected int
}{
{
name: "unset returns the default port",
expected: defaultValue,
},
{
name: "valid port is parsed",
set: true,
value: "9000",
expected: 9000,
},
{
name: "highest port is accepted",
set: true,
value: "65535",
expected: 65535,
},
{
name: "unparseable value is rejected",
set: true,
value: "not-a-port",
expectError: true,
},
{
name: "zero is rejected",
set: true,
value: "0",
expectError: true,
errIs: config.ErrNonPositiveValue,
},
{
name: "above the port range is rejected",
set: true,
value: "65536",
expectError: true,
errIs: config.ErrInvalidPort,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
// Cannot use t.Parallel() here because t.Setenv
// is incompatible with parallel subtests.
if tt.set {
t.Setenv(testEnvKey, tt.value)
} else {
require.NoError(t, os.Unsetenv(testEnvKey))
}
got, err := config.EnvPortForTest(
testEnvKey, defaultValue,
)
if tt.expectError {
require.Error(t, err)
assert.Contains(t, err.Error(), testEnvKey)
if tt.errIs != nil {
require.ErrorIs(t, err, tt.errIs)
}
return
}
require.NoError(t, err)
assert.Equal(t, tt.expected, got)
})
}
}
// buildConfig constructs a Config through fx exactly as the
// application does, returning the config and any construction error.
func buildConfig(t *testing.T) (*config.Config, error) {
t.Helper()
var cfg *config.Config
app := fx.New(
fx.NopLogger,
fx.Provide(
globals.New,
logger.New,
config.New,
),
fx.Populate(&cfg),
)
return cfg, app.Err()
}
func TestNewRejectsBadEnvValues(t *testing.T) {
tests := []struct {
name string
key string
value string
expectError bool
check func(t *testing.T, cfg *config.Config)
}{
{
name: "valid PORT is used",
key: envKeyPort,
value: "9001",
check: func(t *testing.T, cfg *config.Config) {
t.Helper()
assert.Equal(t, 9001, cfg.Port)
},
},
{
name: "unparseable PORT aborts startup",
key: envKeyPort,
value: "eighty-eighty",
expectError: true,
},
{
name: "out-of-range PORT aborts startup",
key: envKeyPort,
value: "70000",
expectError: true,
},
{
name: "valid DEBUG is used",
key: envKeyDebug,
value: "true",
check: func(t *testing.T, cfg *config.Config) {
t.Helper()
assert.True(t, cfg.Debug)
},
},
{
name: "unparseable DEBUG aborts startup",
key: envKeyDebug,
value: "ture",
expectError: true,
},
{
name: "unparseable MAINTENANCE_MODE aborts startup",
key: envKeyMaintenanceMode,
value: "sometimes",
expectError: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
// Cannot use t.Parallel() here because t.Setenv
// is incompatible with parallel subtests.
t.Setenv("WEBHOOKER_ENVIRONMENT", "dev")
t.Setenv(tt.key, tt.value)
cfg, err := buildConfig(t)
if tt.expectError {
require.Error(t, err)
assert.Contains(t, err.Error(), tt.key)
assert.Contains(t, err.Error(), tt.value)
return
}
require.NoError(t, err)
require.NotNil(t, cfg)
tt.check(t, cfg)
})
}
}
// TestNewUsesDefaultsWhenUnset proves the fail-loud behaviour did not
// break the legitimate unset case: absent variables still get their
// documented defaults.
func TestNewUsesDefaultsWhenUnset(t *testing.T) {
t.Setenv("WEBHOOKER_ENVIRONMENT", "dev")
for _, key := range []string{
envKeyPort, envKeyDebug, envKeyMaintenanceMode,
} {
require.NoError(t, os.Unsetenv(key))
}
cfg, err := buildConfig(t)
require.NoError(t, err)
require.NotNil(t, cfg)
assert.Equal(t, 8080, cfg.Port)
assert.False(t, cfg.Debug)
assert.False(t, cfg.MaintenanceMode)
}

View File

@@ -0,0 +1,20 @@
package config
// This file exposes the unexported environment parsing helpers to
// the external config_test package so each helper can be covered by
// its own table-driven test without weakening the package API.
// EnvBoolForTest exposes envBool.
func EnvBoolForTest(key string, defaultValue bool) (bool, error) {
return envBool(key, defaultValue)
}
// EnvPositiveIntForTest exposes envPositiveInt.
func EnvPositiveIntForTest(key string, defaultValue int) (int, error) {
return envPositiveInt(key, defaultValue)
}
// EnvPortForTest exposes envPort.
func EnvPortForTest(key string, defaultValue int) (int, error) {
return envPort(key, defaultValue)
}

View File

@@ -1,229 +0,0 @@
package delivery
import (
"context"
"errors"
"log/slog"
"sync"
"time"
"go.uber.org/fx"
"sneak.berlin/go/webhooker/internal/config"
"sneak.berlin/go/webhooker/internal/database"
"sneak.berlin/go/webhooker/internal/logger"
)
// ArchiveSweeperParams holds the fx dependencies for the
// ArchiveSweeper.
type ArchiveSweeperParams struct {
fx.In
Config *config.Config
Database *database.Database
Engine *Engine
Logger *logger.Logger
}
// ArchiveSweeper periodically prunes expired rows from
// per-webhook archive databases whose database target carries a
// positive expiry.
//
// Without it, pruning happens only when an archive is
// (re)opened, and archives are only ever reopened by writes: an
// archive belonging to a webhook that has stopped receiving
// events would keep its expired rows forever. The sweep closes
// that gap without changing anything for archives whose expiry
// is unset or "never".
//
// It reuses Config.RetentionSweepInterval rather than
// introducing a second interval: this is a retention sweep with
// the same semantics as the event retention reaper.
type ArchiveSweeper struct {
db *database.Database
eng *Engine
log *slog.Logger
interval time.Duration
cancel context.CancelFunc
wg sync.WaitGroup
}
// NewArchiveSweeper creates the archive sweeper and registers
// its fx lifecycle hooks. The background sweep loop starts on
// OnStart and stops cleanly on OnStop via context cancellation.
func NewArchiveSweeper(
lc fx.Lifecycle,
params ArchiveSweeperParams,
) *ArchiveSweeper {
s := &ArchiveSweeper{
db: params.Database,
eng: params.Engine,
log: params.Logger.Get(),
interval: params.Config.RetentionSweepInterval,
}
s.registerHooks(lc)
return s
}
// registerHooks wires the sweeper's start and stop into the fx
// lifecycle. Both hook contexts are deliberately ignored: see
// start for why the background loop must not inherit the start
// hook's context, and stop for why shutdown blocks on the loop
// rather than on the stop hook's deadline.
func (s *ArchiveSweeper) registerHooks(lc fx.Lifecycle) {
lc.Append(fx.Hook{
//nolint:contextcheck // Not passing the hook context is
// the point: see start.
OnStart: func(_ context.Context) error {
s.start()
return nil
},
OnStop: func(_ context.Context) error {
s.stop()
return nil
},
})
}
// start launches the background sweep loop.
//
// The loop's context is derived from context.Background(), NOT
// from the fx OnStart hook context. The hook context carries
// fx's start timeout (15s by default), so a loop derived from it
// is cancelled 15 seconds after the application starts — long
// before the first tick under the default one-hour sweep
// interval, leaving a sweeper that never sweeps. A long-lived
// goroutine must outlive the startup phase, so its lifetime is
// bounded by OnStop instead: stop cancels this context and waits
// on the WaitGroup.
func (s *ArchiveSweeper) start() {
ctx, cancel := context.WithCancel(context.Background())
s.cancel = cancel
s.wg.Add(1)
go s.run(ctx)
s.log.Info(
"archive sweeper started",
"interval", s.interval.String(),
)
}
func (s *ArchiveSweeper) stop() {
s.log.Info("archive sweeper stopping")
if s.cancel != nil {
s.cancel()
}
s.wg.Wait()
s.log.Info("archive sweeper stopped")
}
func (s *ArchiveSweeper) run(ctx context.Context) {
defer s.wg.Done()
ticker := time.NewTicker(s.interval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
s.sweep(ctx)
}
}
}
// sweep prunes every archive whose database target declares a
// positive expiry. Targets belonging to a deleted webhook are
// soft-deleted along with it, so GORM's default scope already
// excludes them.
//
// A failure for one webhook is logged and the sweep continues,
// matching how the write path already treats a prune error as
// non-fatal.
func (s *ArchiveSweeper) sweep(ctx context.Context) {
var targets []database.Target
err := s.db.DB().
Model(&database.Target{}).
Where("type = ?", database.TargetTypeDatabase).
Find(&targets).Error
if err != nil {
s.log.Error(
"archive sweep: failed to list database targets",
"error", err,
)
return
}
for i := range targets {
select {
case <-ctx.Done():
return
default:
}
s.sweepTarget(&targets[i])
}
}
// sweepTarget prunes the archive of a single database target.
// A missing, empty, or "never" expiry parses as a zero duration
// and is skipped entirely, so those archives keep exactly the
// behaviour they had before the sweep existed.
func (s *ArchiveSweeper) sweepTarget(target *database.Target) {
expiry, err := parseArchiveExpiry(target.Config)
if err != nil {
s.log.Error(
"archive sweep: invalid database target config",
"webhook_id", target.WebhookID,
"target_id", target.ID,
"error", err,
)
return
}
if expiry <= 0 {
return
}
if s.eng == nil || s.eng.dbTarget == nil {
return
}
err = s.eng.dbTarget.sweepWebhook(target.WebhookID, expiry)
if err == nil {
return
}
// A writer evicted underneath the sweep means the operator
// deleted the webhook (or its last database target) while the
// sweep was walking the target list. That is an ordinary
// interleaving, not a failure, so it must not produce an
// error line.
if errors.Is(err, errArchiveWriterEvicted) {
s.log.Debug(
"archive sweep: writer evicted mid-sweep",
"webhook_id", target.WebhookID,
"target_id", target.ID,
)
return
}
s.log.Error(
"archive sweep: failed to prune archive",
"webhook_id", target.WebhookID,
"target_id", target.ID,
"error", err,
)
}

View File

@@ -1,930 +0,0 @@
package delivery_test
import (
"context"
"database/sql"
"fmt"
"net/http"
"os"
"path/filepath"
"sync"
"testing"
"time"
"github.com/google/uuid"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"go.uber.org/fx"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
"gorm.io/gorm/clause"
_ "modernc.org/sqlite" // Pure Go SQLite driver.
"sneak.berlin/go/webhooker/internal/database"
"sneak.berlin/go/webhooker/internal/delivery"
)
const (
// sweepRowOld and sweepRowNew are the event ids
// seedArchiveRows assigns to the first and second seeded
// rows.
sweepRowOld = "ev-0"
sweepRowNew = "ev-1"
// sweepConcurrentWrites is how many deliveries the
// concurrent write-plus-sweep test races against the sweep.
sweepConcurrentWrites = 20
)
// sweeperEnv bundles the pieces an archive sweep test drives:
// a main configuration database holding webhooks and targets, a
// delivery engine owning the archive writer registry, and the
// data directory the archive files live in.
type sweeperEnv struct {
sweeper *delivery.ArchiveSweeper
eng *delivery.Engine
mainDB *database.Database
dataDir string
}
func setupSweeperTest(t *testing.T) *sweeperEnv {
t.Helper()
dataDir := t.TempDir()
log := archiveTestLogger()
sqlDB, err := sql.Open(
"sqlite",
fmt.Sprintf(
"file:%s?mode=rwc",
filepath.Join(dataDir, "main.db"),
),
)
require.NoError(t, err)
t.Cleanup(func() { _ = sqlDB.Close() })
gdb, err := gorm.Open(
sqlite.Dialector{Conn: sqlDB}, &gorm.Config{},
)
require.NoError(t, err)
mainDB := database.NewTestDatabase(gdb)
require.NoError(t, mainDB.Migrate())
eng := delivery.NewTestEngineWithDB(
mainDB,
database.NewTestWebhookDBManager(dataDir),
log,
&http.Client{Timeout: 5 * time.Second},
1,
)
return &sweeperEnv{
sweeper: delivery.NewTestArchiveSweeper(
mainDB, eng, log,
),
eng: eng,
mainDB: mainDB,
dataDir: dataDir,
}
}
// archivePath returns where the engine keeps a webhook's
// archive file.
func (env *sweeperEnv) archivePath(webhookID string) string {
return filepath.Join(
env.dataDir, fmt.Sprintf("archive-%s.db", webhookID),
)
}
// seedDatabaseTarget creates a webhook with one database target
// carrying the given target config JSON, and returns the
// webhook id.
func (env *sweeperEnv) seedDatabaseTarget(
t *testing.T, configJSON string,
) string {
t.Helper()
wh := &database.Webhook{
UserID: uuid.New().String(),
Name: "sweep-test",
}
require.NoError(
t,
env.mainDB.DB().
Omit(clause.Associations).
Create(wh).Error,
)
tgt := &database.Target{
WebhookID: wh.ID,
Name: "archive",
Type: database.TargetTypeDatabase,
Active: true,
Config: configJSON,
}
require.NoError(
t,
env.mainDB.DB().
Omit(clause.Associations).
Create(tgt).Error,
)
return wh.ID
}
// seedArchiveRows creates the archive file for a webhook and
// inserts one row per supplied archived-at timestamp, returning
// the archive path. The handle is closed before returning, so
// the archive is idle exactly as it would be with no traffic.
func (env *sweeperEnv) seedArchiveRows(
t *testing.T, webhookID string, archivedAt ...time.Time,
) string {
t.Helper()
path := env.archivePath(webhookID)
sqlDB, err := sql.Open(
"sqlite", fmt.Sprintf("file:%s?mode=rwc", path),
)
require.NoError(t, err)
gdb, err := gorm.Open(
sqlite.Dialector{Conn: sqlDB}, &gorm.Config{},
)
require.NoError(t, err)
require.NoError(
t, gdb.AutoMigrate(&delivery.ExportArchivedEvent{}),
)
for i, at := range archivedAt {
row := delivery.ExportArchivedEvent{
EventID: fmt.Sprintf("ev-%d", i),
WebhookID: webhookID,
Method: http.MethodPost,
Body: `{"seeded":true}`,
ArchivedAt: at,
}
require.NoError(t, gdb.Create(&row).Error)
}
require.NoError(t, sqlDB.Close())
return path
}
// archivedEventIDs returns the event ids currently stored in an
// archive file, read through a separate read-only handle.
func archivedEventIDs(
t *testing.T, path string,
) []string {
t.Helper()
var rows []delivery.ExportArchivedEvent
rdb := openArchiveDBForRead(t, path)
require.NoError(t, rdb.Order("event_id").Find(&rows).Error)
ids := make([]string, 0, len(rows))
for i := range rows {
ids = append(ids, rows[i].EventID)
}
return ids
}
// countArchivedRows counts the rows in an archive file without
// asserting anything, so it is safe to poll from an
// assert.Eventually condition (which runs off the test
// goroutine, where testify assertions must not be used).
func countArchivedRows(path string) (int64, error) {
sqlDB, err := sql.Open(
"sqlite", fmt.Sprintf("file:%s?mode=ro", path),
)
if err != nil {
return 0, err
}
defer func() { _ = sqlDB.Close() }()
gdb, err := gorm.Open(
sqlite.Dialector{Conn: sqlDB}, &gorm.Config{},
)
if err != nil {
return 0, err
}
var count int64
err = gdb.Model(&delivery.ExportArchivedEvent{}).
Count(&count).Error
if err != nil {
return 0, err
}
return count, nil
}
// captureLifecycle is a minimal fx.Lifecycle that records the
// hooks a component registers, so a test can invoke the real
// OnStart/OnStop functions with a context of its choosing.
type captureLifecycle struct {
hooks []fx.Hook
}
func (l *captureLifecycle) Append(h fx.Hook) {
l.hooks = append(l.hooks, h)
}
// TestArchiveSweeper_LoopOutlivesStartHookContext is the
// regression test for a sweeper that never swept. fx calls
// OnStart with a context carrying the application's start
// timeout (15 seconds by default), so a background loop whose
// context is derived from it is cancelled 15 seconds into the
// process — three quarters of an hour before the first tick
// under the default one-hour sweep interval.
//
// The hook context here is already cancelled, which is the same
// defect taken to its limit: a loop that inherits it never runs
// a single tick, while a correctly rooted loop keeps sweeping
// for as long as the process lives. Handing the hook a plain
// context.Background() would assert nothing at all.
func TestArchiveSweeper_LoopOutlivesStartHookContext(
t *testing.T,
) {
t.Parallel()
env := setupSweeperTest(t)
webhookID := env.seedDatabaseTarget(t, `{"expiry":"1h"}`)
now := time.Now()
path := env.seedArchiveRows(
t, webhookID,
now.Add(-48*time.Hour),
now.Add(-time.Minute),
)
env.sweeper.ExportSetInterval(10 * time.Millisecond)
// Drive the genuine fx hooks the application registers,
// rather than a test-only entry point.
lc := &captureLifecycle{}
env.sweeper.ExportRegisterHooks(lc)
require.Len(t, lc.hooks, 1)
hookCtx, cancel := context.WithCancel(context.Background())
cancel()
require.NoError(t, lc.hooks[0].OnStart(hookCtx))
t.Cleanup(func() {
_ = lc.hooks[0].OnStop(context.Background())
})
assert.Eventually(
t,
func() bool {
count, err := countArchivedRows(path)
return err == nil && count == 1
},
5*time.Second,
10*time.Millisecond,
"the sweep loop must keep running after the start "+
"hook's context is done; it pruned nothing, so it "+
"inherited the hook context and died",
)
}
// TestArchiveSweep_DoesNotResurrectEvictedWriter covers the
// interleaving where a sweep tick has already listed a webhook's
// target when the webhook is deleted and its writer evicted. The
// sweep must not put a writer back into the registry: nothing
// would ever evict it again, which is precisely the leak this
// change exists to close.
func TestArchiveSweep_DoesNotResurrectEvictedWriter(
t *testing.T,
) {
t.Parallel()
env := setupSweeperTest(t)
webhookID := env.seedDatabaseTarget(t, `{"expiry":"1h"}`)
env.seedArchiveRows(
t, webhookID, time.Now().Add(-48*time.Hour),
)
// Prime the registry the way a delivery would, then evict as
// the deletion path does. The target row is deliberately left
// in place: this is the tick that listed the webhook before
// the deletion committed.
_, err := env.eng.ExportEnsureArchiveWriter(webhookID)
require.NoError(t, err)
env.eng.EvictWebhook(webhookID)
require.False(t, env.eng.ExportHasArchiveWriter(webhookID))
env.sweeper.ExportSweep(context.Background())
assert.False(
t, env.eng.ExportHasArchiveWriter(webhookID),
"a sweep must never re-register a writer for a webhook "+
"whose registry entry has already been released",
)
}
// TestArchiveSweep_LeavesNoRegistryEntry states the same
// invariant in its general form: sweeping an archive whose
// webhook has no cached writer must not leave one behind, so the
// registry keeps holding only writers a delivery created and an
// eviction can reach.
func TestArchiveSweep_LeavesNoRegistryEntry(t *testing.T) {
t.Parallel()
env := setupSweeperTest(t)
webhookID := env.seedDatabaseTarget(t, `{"expiry":"1h"}`)
path := env.seedArchiveRows(
t, webhookID,
time.Now().Add(-48*time.Hour),
time.Now().Add(-time.Minute),
)
require.False(t, env.eng.ExportHasArchiveWriter(webhookID))
env.sweeper.ExportSweep(context.Background())
assert.Equal(
t, []string{sweepRowNew}, archivedEventIDs(t, path),
"the sweep must still prune an idle archive",
)
assert.False(
t, env.eng.ExportHasArchiveWriter(webhookID),
"the sweep must release the registry entry it created",
)
}
// TestArchiveSweep_KeepsWriterAdoptedByDelivery is the other
// half of that invariant: an entry the sweep created but a
// delivery then claimed belongs to the registry and must survive
// the sweep, or the delivery would be left holding a detached
// writer with an open handle that no eviction can reach.
func TestArchiveSweep_KeepsWriterAdoptedByDelivery(
t *testing.T,
) {
t.Parallel()
env := setupSweeperTest(t)
webhookID := env.seedDatabaseTarget(t, `{"expiry":"1h"}`)
env.seedArchiveRows(
t, webhookID, time.Now().Add(-48*time.Hour),
)
webhookDB := testWebhookDB(t)
event := seedEvent(t, webhookDB, `{"n":1}`)
event.WebhookID = webhookID
d := seedDatabaseTargetDelivery(
t, webhookDB, event, `{"expiry":"1h"}`,
)
env.sweeper.ExportSweep(context.Background())
require.False(t, env.eng.ExportHasArchiveWriter(webhookID))
env.eng.ExportDeliverDatabase(webhookDB, d)
assert.True(
t, env.eng.ExportHasArchiveWriter(webhookID),
"a delivery's writer must stay registered",
)
env.sweeper.ExportSweep(context.Background())
assert.True(
t, env.eng.ExportHasArchiveWriter(webhookID),
"a sweep must not drop a writer a delivery owns",
)
}
// TestArchiveSweep_KeepsWriterAdoptedDuringSweep covers the one
// interleaving the sweepOwned flag exists for, which
// TestArchiveSweep_KeepsWriterAdoptedByDelivery cannot reach: a
// delivery adopting the sweep's own entry WHILE that sweep is
// still running.
//
// The registry operations are driven directly, in the order the
// sweep and a concurrent delivery perform them, so the window is
// exercised deterministically rather than hoped for:
//
// 1. the sweep finds no cached writer and registers one of its
// own, marked sweep-owned;
// 2. a delivery arrives, is handed that very writer, clears the
// flag and opens the archive handle;
// 3. the sweep finishes and releases what it created.
//
// Step 3 must leave the entry alone. Dropping it would detach a
// writer that is holding an open archive handle inside its
// debounce window, and no eviction could ever reach it again —
// exactly the process-lifetime handle leak this change exists to
// close. The eviction at the end proves the entry is still
// reachable.
func TestArchiveSweep_KeepsWriterAdoptedDuringSweep(
t *testing.T,
) {
t.Parallel()
env := setupSweeperTest(t)
webhookID := env.seedDatabaseTarget(t, `{"expiry":"1h"}`)
env.seedArchiveRows(
t, webhookID, time.Now().Add(-48*time.Hour),
)
sweepWriter, created, err := env.eng.ExportSweepWriterFor(
webhookID,
)
require.NoError(t, err)
require.True(
t, created,
"the sweep must have created the registry entry itself",
)
// The delivery lands mid-sweep and adopts the entry.
webhookDB := testWebhookDB(t)
event := seedEvent(t, webhookDB, `{"n":1}`)
event.WebhookID = webhookID
d := seedDatabaseTargetDelivery(
t, webhookDB, event, `{"expiry":"1h"}`,
)
env.eng.ExportDeliverDatabase(webhookDB, d)
adopted := env.eng.ExportArchiveWriterFor(webhookID)
require.NotNil(t, adopted)
require.True(
t, sweepWriter.Same(adopted),
"the delivery must have adopted the sweep's writer",
)
require.True(
t, env.eng.ExportArchiveHandleOpen(webhookID),
"the delivery leaves the archive handle open",
)
// The sweep finishes.
env.eng.ExportReleaseSweepWriter(webhookID, sweepWriter)
require.True(
t, env.eng.ExportHasArchiveWriter(webhookID),
"a writer adopted by a delivery during a sweep must "+
"stay registered, or its open handle is unreachable",
)
env.eng.EvictWebhook(webhookID)
assert.False(
t, env.eng.ExportHasArchiveWriter(webhookID),
"the adopted writer must still be evictable",
)
assert.False(
t, sweepWriter.HandleOpen(),
"eviction must have closed the adopted writer's handle",
)
}
// TestArchiveSweep_ContinuesAfterPerWebhookFailure proves a
// failure for one webhook does not abort the sweep for the
// others: an unparseable expiry and an unreadable archive both
// have to be logged and stepped over.
func TestArchiveSweep_ContinuesAfterPerWebhookFailure(
t *testing.T,
) {
t.Parallel()
env := setupSweeperTest(t)
// Seeded first so the sweep reaches them before the healthy
// webhook: targets come back in insertion order.
badConfigID := env.seedDatabaseTarget(t, `{"expiry":"!!!"}`)
env.seedArchiveRows(
t, badConfigID, time.Now().Add(-48*time.Hour),
)
corruptID := env.seedDatabaseTarget(t, `{"expiry":"1h"}`)
require.NoError(t, os.WriteFile(
env.archivePath(corruptID),
[]byte("this is not a sqlite database"),
0o600,
))
healthyID := env.seedDatabaseTarget(t, `{"expiry":"1h"}`)
healthyPath := env.seedArchiveRows(
t, healthyID,
time.Now().Add(-48*time.Hour),
time.Now().Add(-time.Minute),
)
env.sweeper.ExportSweep(context.Background())
assert.Equal(
t, []string{sweepRowNew},
archivedEventIDs(t, healthyPath),
"a failure for an earlier webhook must not stop the "+
"sweep from pruning the ones after it",
)
}
// TestArchiveSweep_OpenExistingDoesNotCreateFile pins the second
// of the two no-create guards. The first is the stat in
// sweepWebhook; this one is the SQLite open mode, which is what
// protects the window between that stat and the open. Flipping
// the sweep's mode to create-if-missing makes this fail.
func TestArchiveSweep_OpenExistingDoesNotCreateFile(
t *testing.T,
) {
t.Parallel()
dir := t.TempDir()
path := filepath.Join(dir, "archive-absent.db")
w := delivery.NewExportArchiveWriter(
path, archiveTestLogger(), 0,
)
err := w.OpenExisting(time.Hour)
require.Error(
t, err,
"opening a missing archive without create permission "+
"must fail rather than conjure the file",
)
for _, suffix := range archiveFileSuffixes() {
assert.NoFileExists(t, path+suffix)
}
}
// TestArchiveSweep_PrunesIdleArchive is the core regression
// test for this issue: an archive that receives no further
// writes must still lose its expired rows. Before the sweeper
// existed, pruning only ever ran on a write-triggered reopen,
// so an idle archive kept expired rows forever.
func TestArchiveSweep_PrunesIdleArchive(t *testing.T) {
t.Parallel()
env := setupSweeperTest(t)
webhookID := env.seedDatabaseTarget(t, `{"expiry":"1h"}`)
now := time.Now()
path := env.seedArchiveRows(
t, webhookID,
now.Add(-48*time.Hour),
now.Add(-time.Minute),
)
require.Equal(
t, []string{sweepRowOld, sweepRowNew},
archivedEventIDs(t, path),
)
env.sweeper.ExportSweep(context.Background())
assert.Equal(
t, []string{sweepRowNew}, archivedEventIDs(t, path),
"the sweep should prune rows older than the expiry "+
"from an idle archive and keep the rest",
)
}
// TestArchiveSweep_LeavesArchiveClosed proves the sweep does
// not hold the archive open afterwards, so an operator can
// still move the file away for offline retention.
//
// The assertion is made on a writer the test holds a reference
// to, and the handle is proven OPEN before the sweep runs, so the
// test observes the sweep closing it rather than a writer that
// merely never opened anything. Asking the registry instead would
// be vacuous here: the sweep releases an entry it created, and a
// missing entry reports "not open" whether or not anything was
// closed.
func TestArchiveSweep_LeavesArchiveClosed(t *testing.T) {
t.Parallel()
env := setupSweeperTest(t)
webhookID := env.seedDatabaseTarget(t, `{"expiry":"1h"}`)
path := env.seedArchiveRows(
t, webhookID, time.Now().Add(-48*time.Hour),
)
w := delivery.NewExportArchiveWriter(
path, archiveTestLogger(), 0,
)
require.NoError(t, w.OpenExisting(time.Hour))
require.True(
t, w.HandleOpen(),
"the writer must hold an open handle before the sweep",
)
require.NoError(t, w.SweepExpired(time.Hour))
assert.False(
t, w.HandleOpen(),
"an idle archive must end the sweep closed",
)
}
// TestArchiveSweep_ClosesHandleOfRegisteredWriter states the same
// guarantee end to end, through the real sweeper and a writer the
// registry keeps.
//
// The delivery leaves the archive handle open inside its debounce
// window and makes the entry delivery-owned, so the sweep finds a
// cached writer (created is false, nothing is released) and the
// registry query afterwards is answered by a writer that really
// exists. A handle left open here would be doubly wrong: it also
// blocks the operator's move-the-file-away workflow.
func TestArchiveSweep_ClosesHandleOfRegisteredWriter(
t *testing.T,
) {
t.Parallel()
env := setupSweeperTest(t)
webhookID := env.seedDatabaseTarget(t, `{"expiry":"1h"}`)
env.seedArchiveRows(
t, webhookID, time.Now().Add(-48*time.Hour),
)
webhookDB := testWebhookDB(t)
event := seedEvent(t, webhookDB, `{"n":1}`)
event.WebhookID = webhookID
d := seedDatabaseTargetDelivery(
t, webhookDB, event, `{"expiry":"1h"}`,
)
env.eng.ExportDeliverDatabase(webhookDB, d)
require.True(
t, env.eng.ExportArchiveHandleOpen(webhookID),
"the delivery must leave the archive handle open",
)
env.sweeper.ExportSweep(context.Background())
require.True(
t, env.eng.ExportHasArchiveWriter(webhookID),
"the delivery's registry entry must survive the sweep",
)
assert.False(
t, env.eng.ExportArchiveHandleOpen(webhookID),
"the sweep must leave the archive closed",
)
}
// TestArchiveSweep_NeverExpiryUntouched proves the sweep is a
// no-op for the default retention policy, so archives with no
// expiry (or the literal "never") behave exactly as before.
func TestArchiveSweep_NeverExpiryUntouched(t *testing.T) {
t.Parallel()
for _, configJSON := range []string{
`{"expiry":"never"}`,
`{"expiry":""}`,
"",
} {
env := setupSweeperTest(t)
webhookID := env.seedDatabaseTarget(t, configJSON)
path := env.seedArchiveRows(
t, webhookID,
time.Now().Add(-10000*time.Hour),
)
env.sweeper.ExportSweep(context.Background())
assert.Equal(
t, []string{sweepRowOld}, archivedEventIDs(t, path),
"config %q must keep rows forever", configJSON,
)
assert.False(
t, env.eng.ExportHasArchiveWriter(webhookID),
"config %q must leave no registry entry behind",
configJSON,
)
}
}
// TestArchiveSweep_NeverExpirySkipsBeforeOpening pins the
// expiry <= 0 boundary in sweepTarget, which the row assertions
// above cannot reach: pruning is separately gated on a positive
// expiry, so a "never" archive keeps its rows even if the sweep
// does open it.
//
// The spec is stronger than that — a "never" archive is skipped
// before any file is touched — so the archive here exists but has
// never been migrated. Opening it at all would run AutoMigrate
// and create the archive table, which is exactly what must not
// happen.
func TestArchiveSweep_NeverExpirySkipsBeforeOpening(
t *testing.T,
) {
t.Parallel()
env := setupSweeperTest(t)
webhookID := env.seedDatabaseTarget(t, `{"expiry":"never"}`)
path := env.archivePath(webhookID)
seedUnmigratedArchive(t, path)
require.False(t, archiveTableExists(t, path))
env.sweeper.ExportSweep(context.Background())
assert.False(
t, archiveTableExists(t, path),
"a never-expiry archive must not be opened at all",
)
}
// seedUnmigratedArchive creates an archive file that exists but
// carries no archive schema, so any open of it is observable: the
// archive table appears only if something ran AutoMigrate.
func seedUnmigratedArchive(t *testing.T, path string) {
t.Helper()
sqlDB, err := sql.Open(
"sqlite", fmt.Sprintf("file:%s?mode=rwc", path),
)
require.NoError(t, err)
_, err = sqlDB.ExecContext(
t.Context(), "CREATE TABLE placeholder (id INTEGER)",
)
require.NoError(t, err)
require.NoError(t, sqlDB.Close())
}
// archiveTableExists reports whether an archive file has had the
// archive schema migrated into it.
func archiveTableExists(t *testing.T, path string) bool {
t.Helper()
return openArchiveDBForRead(t, path).
Migrator().
HasTable(&delivery.ExportArchivedEvent{})
}
// TestArchiveSweep_DoesNotCreateArchiveFile proves the sweep
// never conjures an archive: a webhook with a database target
// that has never received an event must still have no archive
// file (nor SQLite sidecar) after a sweep.
func TestArchiveSweep_DoesNotCreateArchiveFile(t *testing.T) {
t.Parallel()
env := setupSweeperTest(t)
webhookID := env.seedDatabaseTarget(t, `{"expiry":"1h"}`)
path := env.archivePath(webhookID)
require.NoFileExists(t, path)
env.sweeper.ExportSweep(context.Background())
for _, suffix := range archiveFileSuffixes() {
assert.NoFileExists(
t, path+suffix,
"the sweep must not create an archive file",
)
}
}
// TestArchiveSweep_DoesNotCreateAfterWriterExists covers the
// same guarantee once a writer is cached in the registry but
// the file itself is still absent (for instance because the
// operator moved the archive away).
func TestArchiveSweep_DoesNotCreateAfterWriterExists(
t *testing.T,
) {
t.Parallel()
env := setupSweeperTest(t)
webhookID := env.seedDatabaseTarget(t, `{"expiry":"1h"}`)
path, err := env.eng.ExportEnsureArchiveWriter(webhookID)
require.NoError(t, err)
require.NoFileExists(t, path)
env.sweeper.ExportSweep(context.Background())
assert.NoFileExists(t, path)
}
// TestArchiveSweep_SkipsDeletedWebhookTargets proves that the
// sweep ignores targets soft-deleted along with their webhook,
// so a deleted webhook's archive is never reopened.
func TestArchiveSweep_SkipsDeletedWebhookTargets(t *testing.T) {
t.Parallel()
env := setupSweeperTest(t)
webhookID := env.seedDatabaseTarget(t, `{"expiry":"1h"}`)
path := env.seedArchiveRows(
t, webhookID, time.Now().Add(-48*time.Hour),
)
require.NoError(
t,
env.mainDB.DB().
Where("webhook_id = ?", webhookID).
Delete(&database.Target{}).Error,
)
env.sweeper.ExportSweep(context.Background())
assert.Equal(
t, []string{sweepRowOld}, archivedEventIDs(t, path),
"a deleted target's archive must be left alone",
)
}
// TestArchiveSweep_ConcurrentWrites proves the sweep serialises
// against writes through the per-webhook writer mutex. Run
// under -race, an unsynchronised sweep would be caught here.
func TestArchiveSweep_ConcurrentWrites(t *testing.T) {
t.Parallel()
env := setupSweeperTest(t)
webhookID := env.seedDatabaseTarget(t, `{"expiry":"1h"}`)
webhookDB := testWebhookDB(t)
// The deliveries are seeded up front, on the test's own
// goroutine: the seed helpers assert, and testify assertions
// must not run off the test goroutine.
deliveries := make(
[]*database.Delivery, 0, sweepConcurrentWrites,
)
for range sweepConcurrentWrites {
event := seedEvent(t, webhookDB, `{"n":1}`)
event.WebhookID = webhookID
deliveries = append(
deliveries,
seedDatabaseTargetDelivery(
t, webhookDB, event, `{"expiry":"1h"}`,
),
)
}
var wg sync.WaitGroup
wg.Add(2)
go func() {
defer wg.Done()
for _, d := range deliveries {
env.eng.ExportDeliverDatabase(webhookDB, d)
}
}()
go func() {
defer wg.Done()
for range sweepConcurrentWrites {
env.sweeper.ExportSweep(context.Background())
}
}()
wg.Wait()
assert.FileExists(t, env.archivePath(webhookID))
}
// TestArchiveSweeper_StopsCleanly proves the background loop
// exits on OnStop rather than leaking a goroutine.
func TestArchiveSweeper_StopsCleanly(t *testing.T) {
t.Parallel()
env := setupSweeperTest(t)
webhookID := env.seedDatabaseTarget(t, `{"expiry":"1h"}`)
env.seedArchiveRows(
t, webhookID, time.Now().Add(-48*time.Hour),
)
env.sweeper.ExportSetInterval(time.Millisecond)
env.sweeper.ExportStart()
// stop blocks on the loop's WaitGroup, so returning at all
// proves the loop observed the cancellation and exited.
env.sweeper.ExportStop()
}

View File

@@ -94,23 +94,6 @@ type Notifier interface {
Notify(tasks []Task) Notify(tasks []Task)
} }
// WebhookEvictor releases the delivery engine's per-webhook
// state for a webhook that no longer needs it — currently the
// cached archive writer of the database target, whose open
// file handle would otherwise outlive the webhook.
//
// It is deliberately separate from Notifier and deliberately
// one method wide: archiving lifecycle is not notification, and
// a single-method interface keeps the handlers package free of
// any dependency on the engine's internals while staying
// trivially fakeable in tests.
//
// EvictWebhook never deletes an archive file. It is idempotent
// and is a no-op for a webhook with no engine state.
type WebhookEvictor interface {
EvictWebhook(webhookID string)
}
// EngineParams are the fx dependencies for the delivery // EngineParams are the fx dependencies for the delivery
// engine. // engine.
type EngineParams struct { type EngineParams struct {
@@ -144,10 +127,6 @@ type Engine struct {
// httpTarget is retained so tests can reach the HTTP // httpTarget is retained so tests can reach the HTTP
// target's shared client and circuit breakers. // target's shared client and circuit breakers.
httpTarget *httpTarget httpTarget *httpTarget
// dbTarget is retained so the engine can reach the archive
// writer registry for webhook eviction and the idle sweep.
dbTarget *databaseTarget
} }
// New creates and registers the delivery engine with the // New creates and registers the delivery engine with the
@@ -203,19 +182,6 @@ func (e *Engine) Notify(tasks []Task) {
} }
} }
// EvictWebhook implements WebhookEvictor. It releases the
// engine's per-webhook archiving state: the database target's
// cached archive writer is dropped from the registry and its
// file handle closed. The archive file itself is left on disk
// — it is long-term storage the operator owns.
func (e *Engine) EvictWebhook(webhookID string) {
if e.dbTarget == nil {
return
}
e.dbTarget.evict(webhookID)
}
// ScheduleRetry schedules a task to be re-enqueued onto the // ScheduleRetry schedules a task to be re-enqueued onto the
// retry channel after delay. It implements the Scheduler // retry channel after delay. It implements the Scheduler
// interface the targets use to own their durable retries. // interface the targets use to own their durable retries.

View File

@@ -7,17 +7,10 @@ import (
"net/http" "net/http"
"time" "time"
"go.uber.org/fx"
"gorm.io/gorm" "gorm.io/gorm"
"sneak.berlin/go/webhooker/internal/database" "sneak.berlin/go/webhooker/internal/database"
) )
// ErrExportArchiveWriterEvicted exposes the sentinel returned by
// an evicted archive writer. It carries the Err prefix rather
// than this file's usual Export one because it is a sentinel
// error.
var ErrExportArchiveWriterEvicted = errArchiveWriterEvicted
// Exported constants for test access. // Exported constants for test access.
const ( const (
ExportDeliveryChannelSize = deliveryChannelSize ExportDeliveryChannelSize = deliveryChannelSize
@@ -335,183 +328,6 @@ func (e *ExportArchiveWriter) DB() *gorm.DB {
return e.w.db return e.w.db
} }
// Path returns the archive file the writer owns.
func (e *ExportArchiveWriter) Path() string {
return e.w.path
}
// OpenExisting opens the archive without permitting creation,
// the way the idle sweep does.
func (e *ExportArchiveWriter) OpenExisting(
expiry time.Duration,
) error {
return e.w.openMode(archiveModeExisting, expiry)
}
// SweepExpired runs an idle sweep of the archive.
func (e *ExportArchiveWriter) SweepExpired(
expiry time.Duration,
) error {
return e.w.sweepExpired(expiry)
}
// Evict marks the writer evicted and closes its handle, exactly
// as leaving the registry does.
func (e *ExportArchiveWriter) Evict() {
e.w.evict()
}
// HandleOpen reports whether the writer currently holds an open
// archive handle.
func (e *ExportArchiveWriter) HandleOpen() bool {
e.w.mu.Lock()
defer e.w.mu.Unlock()
return e.w.db != nil
}
// Same reports whether both wrappers refer to the very same
// underlying archive writer, so a test can prove a registry entry
// is the writer it was handed rather than a replacement.
func (e *ExportArchiveWriter) Same(
other *ExportArchiveWriter,
) bool {
return other != nil && e.w == other.w
}
// ExportArchiveWriterFor returns the archive writer the registry
// currently caches for a webhook, or nil when none is cached. It
// never creates one, so a test can hold a reference to the very
// writer an eviction is about to detach.
func (e *Engine) ExportArchiveWriterFor(
webhookID string,
) *ExportArchiveWriter {
e.dbTarget.mu.Lock()
defer e.dbTarget.mu.Unlock()
w, ok := e.dbTarget.writers[webhookID]
if !ok {
return nil
}
return &ExportArchiveWriter{w: w}
}
// ExportHasArchiveWriter reports whether the database target
// currently caches an archive writer for a webhook.
func (e *Engine) ExportHasArchiveWriter(
webhookID string,
) bool {
e.dbTarget.mu.Lock()
defer e.dbTarget.mu.Unlock()
_, ok := e.dbTarget.writers[webhookID]
return ok
}
// ExportArchiveHandleOpen reports whether the cached archive
// writer for a webhook holds an open database handle. It
// returns false when no writer is cached.
func (e *Engine) ExportArchiveHandleOpen(
webhookID string,
) bool {
e.dbTarget.mu.Lock()
w, ok := e.dbTarget.writers[webhookID]
e.dbTarget.mu.Unlock()
if !ok {
return false
}
w.mu.Lock()
defer w.mu.Unlock()
return w.db != nil
}
// ExportEnsureArchiveWriter creates (if needed) and returns the
// archive file path of the cached writer for a webhook, so a
// test can prime the registry the way a delivery would.
func (e *Engine) ExportEnsureArchiveWriter(
webhookID string,
) (string, error) {
w, err := e.dbTarget.writerFor(webhookID)
if err != nil {
return "", err
}
return w.path, nil
}
// ExportSweepWriterFor takes a webhook's registry writer exactly
// as the idle sweep does, reporting whether the sweep had to
// create the entry. It lets a test drive the registry through the
// sweep's own entry point instead of choreographing goroutines.
func (e *Engine) ExportSweepWriterFor(
webhookID string,
) (*ExportArchiveWriter, bool, error) {
w, created, err := e.dbTarget.sweepWriterFor(webhookID)
if err != nil {
return nil, false, err
}
return &ExportArchiveWriter{w: w}, created, nil
}
// ExportReleaseSweepWriter releases a sweep-created registry entry
// exactly as a finished sweep does.
func (e *Engine) ExportReleaseSweepWriter(
webhookID string, w *ExportArchiveWriter,
) {
e.dbTarget.releaseSweepWriter(webhookID, w.w)
}
// NewTestArchiveSweeper builds an ArchiveSweeper backed by the
// given main database and engine, without the fx lifecycle.
// Intended for tests.
func NewTestArchiveSweeper(
db *database.Database,
eng *Engine,
log *slog.Logger,
) *ArchiveSweeper {
return &ArchiveSweeper{
db: db,
eng: eng,
log: log,
interval: time.Hour,
}
}
// ExportSweep runs a single archive sweep synchronously for
// tests.
func (s *ArchiveSweeper) ExportSweep(ctx context.Context) {
s.sweep(ctx)
}
// ExportStart starts the sweeper's background loop for tests.
func (s *ArchiveSweeper) ExportStart() {
s.start()
}
// ExportRegisterHooks registers the sweeper's real fx lifecycle
// hooks on a lifecycle supplied by a test, so a test can drive
// the exact OnStart/OnStop functions the application runs and
// hand OnStart the kind of context fx actually supplies.
func (s *ArchiveSweeper) ExportRegisterHooks(lc fx.Lifecycle) {
s.registerHooks(lc)
}
// ExportStop stops the sweeper's background loop for tests.
func (s *ArchiveSweeper) ExportStop() {
s.stop()
}
// ExportSetInterval overrides the sweep interval for tests.
func (s *ArchiveSweeper) ExportSetInterval(d time.Duration) {
s.interval = d
}
// ExportParseArchiveExpiry exposes parseArchiveExpiry. // ExportParseArchiveExpiry exposes parseArchiveExpiry.
func ExportParseArchiveExpiry( func ExportParseArchiveExpiry(
configJSON string, configJSON string,

View File

@@ -90,15 +90,12 @@ func (e *Engine) initTargets(client *http.Client) {
client: client, client: client,
} }
dbT := &databaseTarget{eng: e}
e.httpTarget = httpT e.httpTarget = httpT
e.dbTarget = dbT
e.targets = map[database.TargetType]Target{ e.targets = map[database.TargetType]Target{
database.TargetTypeHTTP: httpT, database.TargetTypeHTTP: httpT,
database.TargetTypeSlack: slackT, database.TargetTypeSlack: slackT,
database.TargetTypeDatabase: dbT, database.TargetTypeDatabase: &databaseTarget{eng: e},
database.TargetTypeLog: &logTarget{eng: e}, database.TargetTypeLog: &logTarget{eng: e},
} }
} }

View File

@@ -5,7 +5,6 @@ import (
"fmt" "fmt"
"path/filepath" "path/filepath"
"sync" "sync"
"time"
"gorm.io/gorm" "gorm.io/gorm"
"sneak.berlin/go/webhooker/internal/database" "sneak.berlin/go/webhooker/internal/database"
@@ -112,184 +111,27 @@ func (t *databaseTarget) archive(d *database.Delivery) error {
func (t *databaseTarget) writerFor( func (t *databaseTarget) writerFor(
webhookID string, webhookID string,
) (*archiveWriter, error) { ) (*archiveWriter, error) {
path, err := t.archivePath(webhookID)
if err != nil {
return nil, err
}
t.mu.Lock()
defer t.mu.Unlock()
if t.writers == nil {
t.writers = make(map[string]*archiveWriter)
}
w, ok := t.writers[webhookID]
if !ok {
w = newArchiveWriter(path, t.eng.log)
t.writers[webhookID] = w
}
// A delivery claims the entry: even if the idle sweep created
// it moments ago, it now belongs to the registry proper and
// the sweep must leave it in place when it finishes.
w.sweepOwned = false
return w, nil
}
// sweepWriterFor returns the archive writer the idle sweep should
// prune a webhook through, together with whether the sweep itself
// created the registry entry.
//
// The sweep must route its prune through the registered writer so
// the writer's mutex orders it against concurrent writes, but it
// must never leave a registry entry behind: a sweep that ran
// concurrently with the webhook's deletion would otherwise
// re-create an entry that nothing will ever evict again, which is
// exactly the leak eviction exists to prevent. An entry the sweep
// creates is therefore marked sweep-owned and handed back to
// releaseSweepWriter when the sweep is done.
func (t *databaseTarget) sweepWriterFor(
webhookID string,
) (*archiveWriter, bool, error) {
path, err := t.archivePath(webhookID)
if err != nil {
return nil, false, err
}
t.mu.Lock()
defer t.mu.Unlock()
if t.writers == nil {
t.writers = make(map[string]*archiveWriter)
}
w, ok := t.writers[webhookID]
if ok {
return w, false, nil
}
w = newArchiveWriter(path, t.eng.log)
w.sweepOwned = true
t.writers[webhookID] = w
return w, true, nil
}
// releaseSweepWriter drops a registry entry that the idle sweep
// created, so a sweep leaves the registry exactly as it found it.
//
// The entry is removed only if it is still the very writer the
// sweep installed and no delivery has claimed it in the meantime
// (writerFor clears sweepOwned when it hands a writer to the
// write path). Both conditions are evaluated under the registry
// lock, so an eviction that raced the sweep — which removes the
// entry outright — simply finds nothing left to do here, and a
// delivery that adopted the writer keeps a registered, evictable
// one.
func (t *databaseTarget) releaseSweepWriter(
webhookID string, w *archiveWriter,
) {
t.mu.Lock()
defer t.mu.Unlock()
cur, ok := t.writers[webhookID]
if !ok || cur != w || !cur.sweepOwned {
return
}
delete(t.writers, webhookID)
}
// archivePath returns the archive file path for a webhook: it
// lives beside the per-webhook event database in the data
// directory. It does not touch the filesystem.
func (t *databaseTarget) archivePath(
webhookID string,
) (string, error) {
if t.eng.dbManager == nil { if t.eng.dbManager == nil {
return "", errArchiveNoDataDir return nil, errArchiveNoDataDir
} }
dir := filepath.Dir(t.eng.dbManager.DBPath(webhookID)) dir := filepath.Dir(t.eng.dbManager.DBPath(webhookID))
path := filepath.Join(
return filepath.Join(
dir, fmt.Sprintf("archive-%s.db", webhookID), dir, fmt.Sprintf("archive-%s.db", webhookID),
), nil )
}
// evict drops a webhook's archive writer from the registry and
// closes its handle, so a deleted webhook does not leave a
// writer (and an open archive handle within its debounce
// window) alive for the process lifetime.
//
// The map entry is removed under the registry lock, which is
// then released before the handle is closed under the writer's
// own lock: that ordering keeps the registry available to other
// webhooks while an in-flight write on this one drains, and
// closing under the writer's lock means eviction can never race
// a write.
//
// Eviction is idempotent and silent for a webhook with no
// writer, which is the common case: a webhook with no database
// target never creates one. It never deletes the archive file.
func (t *databaseTarget) evict(webhookID string) {
t.mu.Lock() t.mu.Lock()
defer t.mu.Unlock()
if t.writers == nil {
t.writers = make(map[string]*archiveWriter)
}
w, ok := t.writers[webhookID] w, ok := t.writers[webhookID]
if ok {
delete(t.writers, webhookID)
}
t.mu.Unlock()
if !ok { if !ok {
return w = newArchiveWriter(path, t.eng.log)
t.writers[webhookID] = w
} }
w.evict() return w, nil
t.eng.log.Info(
"evicted archive writer",
"webhook_id", webhookID,
"path", w.path,
)
}
// sweepWebhook prunes one webhook's archive of rows older than
// expiry, without requiring a write. It returns nil (nothing to
// do) when the archive file does not exist, so a sweep never
// creates an archive for a webhook that has a database target
// but has never received an event.
//
// It also never leaves a registry entry behind: an entry it had
// to create to reach the writer's mutex is released again once
// the prune is done, so a sweep racing a webhook deletion cannot
// resurrect the writer the eviction just dropped.
func (t *databaseTarget) sweepWebhook(
webhookID string, expiry time.Duration,
) error {
path, err := t.archivePath(webhookID)
if err != nil {
return err
}
// Check before taking a writer at all: a webhook whose
// archive has never been created gets no writer, no handle,
// and no file.
if !fileExists(path) {
return nil
}
w, created, err := t.sweepWriterFor(webhookID)
if err != nil {
return err
}
if created {
defer t.releaseSweepWriter(webhookID, w)
}
return w.sweepExpired(expiry)
} }

View File

@@ -24,20 +24,6 @@ const archiveExpiryNever = "never"
// offline archiving, but never more than once per this window. // offline archiving, but never more than once per this window.
const archiveReopenDebounce = time.Second const archiveReopenDebounce = time.Second
const (
// archiveModeCreate is the SQLite URI mode used by the write
// path: open the archive file, creating it if missing, so a
// first write (or a write after the operator moved the file
// away) recreates it.
archiveModeCreate = "rwc"
// archiveModeExisting is the SQLite URI mode used by the idle
// sweep: open read-write but never create. A sweep must never
// conjure an empty archive file for a webhook that has a
// database target but has never received an event.
archiveModeExisting = "rw"
)
var ( var (
// errArchiveMissingWebhookID is returned when an event to // errArchiveMissingWebhookID is returned when an event to
// archive has no webhook id to key its archive file on. // archive has no webhook id to key its archive file on.
@@ -58,15 +44,6 @@ var (
errArchiveExpiryNotPositive = errors.New( errArchiveExpiryNotPositive = errors.New(
"expiry must be a positive duration or \"never\"", "expiry must be a positive duration or \"never\"",
) )
// errArchiveWriterEvicted is returned when a writer that has
// been evicted (its webhook was deleted, or its last database
// target was removed) is used again. An evicted writer is no
// longer in the registry, so reopening its file would leak a
// handle nothing owns.
errArchiveWriterEvicted = errors.New(
"archive writer has been evicted",
)
) )
// databaseTargetConfig is the optional per-target JSON config // databaseTargetConfig is the optional per-target JSON config
@@ -184,25 +161,6 @@ type archiveWriter struct {
db *gorm.DB db *gorm.DB
lastReopen time.Time lastReopen time.Time
reopens int reopens int
// evicted marks a writer that has been removed from the
// per-webhook registry. Its handle is closed and it must
// never open the file again: nothing holds it any more, so a
// reopen would leak the handle for the process lifetime.
evicted bool
// sweepOwned marks a registry entry that the idle sweep
// created because no writer was cached for the webhook. The
// sweep removes such an entry again when it is done, so a
// sweep can never leave — or resurrect — a registry entry
// for a webhook that has been deleted. A delivery that adopts
// the writer clears the flag, handing the entry to the
// registry proper.
//
// Unlike every other field here it is guarded by
// databaseTarget.mu, not by this writer's mu: it describes the
// registry entry rather than the file.
sweepOwned bool
} }
// newArchiveWriter builds an archiveWriter for a file path with // newArchiveWriter builds an archiveWriter for a file path with
@@ -227,12 +185,6 @@ func (w *archiveWriter) write(
w.mu.Lock() w.mu.Lock()
defer w.mu.Unlock() defer w.mu.Unlock()
if w.evicted {
return fmt.Errorf(
"%w: %s", errArchiveWriterEvicted, w.path,
)
}
if w.db == nil || !fileExists(w.path) { if w.db == nil || !fileExists(w.path) {
err := w.reopen(expiry) err := w.reopen(expiry)
if err != nil { if err != nil {
@@ -260,19 +212,7 @@ func (w *archiveWriter) write(
// its schema, records the reopen time, and prunes expired rows // its schema, records the reopen time, and prunes expired rows
// when expiry is positive. // when expiry is positive.
func (w *archiveWriter) open(expiry time.Duration) error { func (w *archiveWriter) open(expiry time.Duration) error {
return w.openMode(archiveModeCreate, expiry) dbURL := fmt.Sprintf("file:%s?mode=rwc", w.path)
}
// openMode opens the archive file with the given SQLite URI
// mode, migrates its schema, records the reopen time, and
// prunes expired rows when expiry is positive. The write path
// passes archiveModeCreate so a missing file is recreated; the
// idle sweep passes archiveModeExisting so a missing file is an
// error rather than a newly conjured empty archive.
func (w *archiveWriter) openMode(
mode string, expiry time.Duration,
) error {
dbURL := fmt.Sprintf("file:%s?mode=%s", w.path, mode)
sqlDB, err := sql.Open("sqlite", dbURL) sqlDB, err := sql.Open("sqlite", dbURL)
if err != nil { if err != nil {
@@ -335,70 +275,11 @@ func (w *archiveWriter) close() {
w.db = nil w.db = nil
} }
// sweepExpired prunes an archive that may have gone idle, with
// no write to trigger the usual on-reopen prune. It takes the
// writer's own mutex for the whole operation, so a sweep is
// ordered against concurrent writes rather than reaching around
// them to the file.
//
// It never creates the archive file: a missing file is skipped,
// and the reopen uses archiveModeExisting so SQLite itself
// refuses to create one if the file disappears between the
// check and the open.
//
// The archive is left CLOSED afterwards. An idle archive holding
// no handle is what keeps the operator's move-the-file-away
// workflow working; the next write reopens (and recreates) the
// file as it always has.
func (w *archiveWriter) sweepExpired(expiry time.Duration) error {
w.mu.Lock()
defer w.mu.Unlock()
if w.evicted {
return fmt.Errorf(
"%w: %s", errArchiveWriterEvicted, w.path,
)
}
if !fileExists(w.path) {
return nil
}
// Drop any live handle first so the prune runs against a
// freshly opened file, matching the write path's semantics.
w.close()
err := w.openMode(archiveModeExisting, expiry)
if err != nil {
return err
}
w.close()
return nil
}
// evict closes the writer's handle and marks it unusable. It is
// called when the writer leaves the registry, either because the
// webhook was deleted or because its last database target was
// removed. The archive FILE is deliberately left on disk: it is
// long-term storage an operator may still want.
func (w *archiveWriter) evict() {
w.mu.Lock()
defer w.mu.Unlock()
w.evicted = true
w.close()
}
// prune deletes archived rows older than expiry, measured from // prune deletes archived rows older than expiry, measured from
// each row's archived time. It runs on every (re)open, so a // each row's archived time. It runs on every (re)open, and
// steadily written archive is swept by its own write traffic. An // because the file is reopened after writes this keeps the
// archive that goes idle receives no further reopens, which is // archive swept without a separate background sweeper. Failures
// why ArchiveSweeper exists to drive sweepExpired on a timer. // are logged, not fatal: a prune error must not stop archiving.
// Failures are logged, not fatal: a prune error must not stop
// archiving.
func (w *archiveWriter) prune(expiry time.Duration) { func (w *archiveWriter) prune(expiry time.Duration) {
cutoff := time.Now().Add(-expiry) cutoff := time.Now().Add(-expiry)

View File

@@ -1,363 +0,0 @@
package delivery_test
import (
"errors"
"fmt"
"net/http"
"os"
"path/filepath"
"sync"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"sneak.berlin/go/webhooker/internal/database"
"sneak.berlin/go/webhooker/internal/delivery"
)
// evictTestEngine builds an engine backed by a temporary data
// directory and returns it along with that directory.
func evictTestEngine(t *testing.T) (*delivery.Engine, string) {
t.Helper()
dataDir := t.TempDir()
eng := delivery.NewTestEngineWithDB(
nil,
database.NewTestWebhookDBManager(dataDir),
archiveTestLogger(),
&http.Client{Timeout: 5 * time.Second},
1,
)
return eng, dataDir
}
// TestEvictWebhook_ClosesAndRemovesWriter proves that evicting
// a webhook drops its archive writer from the registry and
// closes the open archive handle, rather than leaving both
// alive for the process lifetime.
func TestEvictWebhook_ClosesAndRemovesWriter(t *testing.T) {
t.Parallel()
eng, dataDir := evictTestEngine(t)
webhookDB := testWebhookDB(t)
event := seedEvent(t, webhookDB, `{"archived":true}`)
d := seedDatabaseTargetDelivery(t, webhookDB, event, "")
eng.ExportDeliverDatabase(webhookDB, d)
webhookID := event.WebhookID
require.True(
t, eng.ExportHasArchiveWriter(webhookID),
"a delivery should have cached an archive writer",
)
require.True(
t, eng.ExportArchiveHandleOpen(webhookID),
"the writer should hold an open handle after a write",
)
eng.EvictWebhook(webhookID)
assert.False(
t, eng.ExportHasArchiveWriter(webhookID),
"eviction should remove the registry entry",
)
assert.False(
t, eng.ExportArchiveHandleOpen(webhookID),
"eviction should close the archive handle",
)
archivePath := filepath.Join(
dataDir, fmt.Sprintf("archive-%s.db", webhookID),
)
assert.FileExists(
t, archivePath,
"eviction must not delete the archive file",
)
}
// TestEvictWebhook_UnknownWebhookIsNoOp proves eviction is safe
// for the common case of a webhook that never had a database
// target, and that repeating it does not panic.
func TestEvictWebhook_UnknownWebhookIsNoOp(t *testing.T) {
t.Parallel()
eng, _ := evictTestEngine(t)
assert.NotPanics(t, func() {
eng.EvictWebhook("no-such-webhook")
eng.EvictWebhook("no-such-webhook")
})
assert.False(
t, eng.ExportHasArchiveWriter("no-such-webhook"),
"eviction must not create a writer",
)
}
// evictTestRow builds an archive row for the eviction tests.
func evictTestRow(eventID string) delivery.ExportArchivedEvent {
return delivery.ExportArchivedEvent{
EventID: eventID,
WebhookID: "wh-evict",
Method: http.MethodPost,
Body: `{"seeded":true}`,
}
}
// TestEvictedWriter_WriteDoesNotReopenFile is the direct test of
// the evicted guard on the write path. A writer that has left
// the registry is held by nobody, so a handle it opened could
// never be closed again: it must refuse the write outright
// rather than recreate the archive behind the registry's back.
//
// The archive file is removed before the eviction, so an
// unguarded write is unmistakable — it recreates the file.
func TestEvictedWriter_WriteDoesNotReopenFile(t *testing.T) {
t.Parallel()
path := filepath.Join(t.TempDir(), "archive-evicted.db")
w := delivery.NewExportArchiveWriter(
path, archiveTestLogger(), 0,
)
require.NoError(t, w.Write(evictTestRow("ev-1"), 0))
require.FileExists(t, path)
// The operator moves the archive away for offline retention,
// which the write path would ordinarily undo on the next
// write by recreating the file.
require.NoError(t, os.Remove(path))
w.Evict()
err := w.Write(evictTestRow("ev-2"), 0)
require.ErrorIs(
t, err, delivery.ErrExportArchiveWriterEvicted,
"an evicted writer must refuse writes",
)
assert.NoFileExists(
t, path,
"an evicted writer must not reopen (or recreate) the "+
"archive file",
)
assert.False(
t, w.HandleOpen(),
"an evicted writer must hold no handle",
)
}
// TestEvictedWriter_SweepDoesNotReopenFile is the same test for
// the sweep path: an idle sweep that reaches a writer already
// evicted underneath it must return the sentinel rather than
// reopen a file nothing owns.
func TestEvictedWriter_SweepDoesNotReopenFile(t *testing.T) {
t.Parallel()
path := filepath.Join(t.TempDir(), "archive-evicted.db")
w := delivery.NewExportArchiveWriter(
path, archiveTestLogger(), 0,
)
require.NoError(t, w.Write(evictTestRow("ev-1"), 0))
require.FileExists(t, path)
w.Evict()
err := w.SweepExpired(time.Hour)
require.ErrorIs(
t, err, delivery.ErrExportArchiveWriterEvicted,
"an evicted writer must refuse an idle sweep",
)
assert.False(
t, w.HandleOpen(),
"a refused sweep must not leave a handle open",
)
}
// racingWrites drives a pack of goroutines writing to one
// archive writer until each is refused, so an eviction on the
// test goroutine has to take the writer's mutex away from writes
// that are already contending for it.
type racingWrites struct {
wg sync.WaitGroup
mu sync.Mutex
sawEvicted bool
otherErr error
started chan struct{}
}
// racingWriteGoroutines is how many goroutines contend for the
// writer's mutex while the eviction lands.
const racingWriteGoroutines = 4
// startRacingWrites launches the writing goroutines. Each writes
// in a loop and stops at its first error, recording whether that
// error was the eviction sentinel. The deadline is a backstop
// against a hang, not a timing assumption: the first write after
// the eviction is refused.
func startRacingWrites(
w *delivery.ExportArchiveWriter,
) *racingWrites {
r := &racingWrites{
started: make(chan struct{}, racingWriteGoroutines),
}
deadline := time.Now().Add(10 * time.Second)
r.wg.Add(racingWriteGoroutines)
for i := range racingWriteGoroutines {
go func() {
defer r.wg.Done()
first := true
for time.Now().Before(deadline) {
err := w.Write(
evictTestRow(fmt.Sprintf("ev-%d", i)), 0,
)
if first {
r.started <- struct{}{}
first = false
}
if err == nil {
continue
}
r.record(err)
return
}
}()
}
return r
}
// record classifies the error that stopped one goroutine.
func (r *racingWrites) record(err error) {
r.mu.Lock()
defer r.mu.Unlock()
if errors.Is(err, delivery.ErrExportArchiveWriterEvicted) {
r.sawEvicted = true
return
}
r.otherErr = err
}
// awaitFirstWrite blocks until at least one write has run, so
// the eviction that follows is a genuine race.
func (r *racingWrites) awaitFirstWrite() {
<-r.started
}
// wait joins the goroutines and reports whether any write was
// refused with the eviction sentinel, plus any unexpected error.
func (r *racingWrites) wait() (bool, error) {
r.wg.Wait()
r.mu.Lock()
defer r.mu.Unlock()
return r.sawEvicted, r.otherErr
}
// TestEvictWebhook_RacingWriteDoesNotReopenHandle exercises the
// interleaving the evicted flag exists for: writes already
// contending for the writer's mutex when the eviction takes it.
// The write that wins the mutex after the eviction must abandon
// its work rather than reopen the archive, leaving the writer
// permanently handle-free. Run under -race.
func TestEvictWebhook_RacingWriteDoesNotReopenHandle(
t *testing.T,
) {
t.Parallel()
eng, _ := evictTestEngine(t)
webhookDB := testWebhookDB(t)
event := seedEvent(t, webhookDB, `{"archived":true}`)
d := seedDatabaseTargetDelivery(t, webhookDB, event, "")
// Prime the registry so the test can hold the very writer the
// eviction is about to detach.
eng.ExportDeliverDatabase(webhookDB, d)
w := eng.ExportArchiveWriterFor(event.WebhookID)
require.NotNil(t, w)
require.True(t, w.HandleOpen())
race := startRacingWrites(w)
// Evict only once writes are genuinely in flight, so the
// eviction has to contend for the writer's mutex.
race.awaitFirstWrite()
eng.EvictWebhook(event.WebhookID)
sawEvicted, otherErr := race.wait()
require.NoError(t, otherErr)
assert.True(
t, sawEvicted,
"a write after eviction must be refused",
)
assert.False(
t, w.HandleOpen(),
"no write may reopen the archive once the writer has "+
"been evicted",
)
assert.False(
t, eng.ExportHasArchiveWriter(event.WebhookID),
"the registry entry must stay gone",
)
}
// TestEvictWebhook_LaterDeliveryRecreatesWriter proves eviction
// does not break archiving for a webhook that is still alive: a
// subsequent delivery gets a brand new writer from the registry.
// It says nothing about the evicted writer itself — that is what
// TestEvictedWriter_WriteDoesNotReopenFile covers.
func TestEvictWebhook_LaterDeliveryRecreatesWriter(t *testing.T) {
t.Parallel()
eng, _ := evictTestEngine(t)
webhookDB := testWebhookDB(t)
event := seedEvent(t, webhookDB, `{"archived":true}`)
d := seedDatabaseTargetDelivery(t, webhookDB, event, "")
eng.ExportDeliverDatabase(webhookDB, d)
require.True(
t, eng.ExportHasArchiveWriter(event.WebhookID),
)
eng.EvictWebhook(event.WebhookID)
// A fresh delivery for the same webhook gets a brand new
// writer from the registry, so archiving keeps working.
second := seedDatabaseTargetDelivery(
t, webhookDB, event, "",
)
eng.ExportDeliverDatabase(webhookDB, second)
assert.True(
t, eng.ExportHasArchiveWriter(event.WebhookID),
"a later delivery should recreate the writer",
)
}

View File

@@ -50,13 +50,6 @@ func openArchiveDBForRead(
return gdb return gdb
} }
// archiveFileSuffixes returns the archive file itself and the
// SQLite sidecars that accompany an open database. A test that
// asserts no archive was created has to check all of them.
func archiveFileSuffixes() []string {
return []string{"", "-wal", "-shm"}
}
// removeArchiveFiles simulates an operator moving the archive // removeArchiveFiles simulates an operator moving the archive
// away by deleting the SQLite file and its sidecar files. // away by deleting the SQLite file and its sidecar files.
func removeArchiveFiles(t *testing.T, path string) { func removeArchiveFiles(t *testing.T, path string) {

View File

@@ -51,7 +51,6 @@ type HandlersParams struct {
Healthcheck *healthcheck.Healthcheck Healthcheck *healthcheck.Healthcheck
Session *session.Session Session *session.Session
Notifier delivery.Notifier Notifier delivery.Notifier
Evictor delivery.WebhookEvictor
} }
// Handlers provides HTTP handler methods for all application // Handlers provides HTTP handler methods for all application
@@ -64,7 +63,6 @@ type Handlers struct {
dbMgr *database.WebhookDBManager dbMgr *database.WebhookDBManager
session *session.Session session *session.Session
notifier delivery.Notifier notifier delivery.Notifier
evictor delivery.WebhookEvictor
templates map[string]*template.Template templates map[string]*template.Template
} }
@@ -99,7 +97,6 @@ func New(
s.dbMgr = params.WebhookDBMgr s.dbMgr = params.WebhookDBMgr
s.session = params.Session s.session = params.Session
s.notifier = params.Notifier s.notifier = params.Notifier
s.evictor = params.Evictor
// Parse all page templates once at startup // Parse all page templates once at startup
s.templates = map[string]*template.Template{ s.templates = map[string]*template.Template{

View File

@@ -4,7 +4,6 @@ import (
"context" "context"
"net/http" "net/http"
"net/http/httptest" "net/http/httptest"
"sync"
"testing" "testing"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
@@ -25,32 +24,6 @@ type noopNotifier struct{}
func (n *noopNotifier) Notify([]delivery.Task) {} func (n *noopNotifier) Notify([]delivery.Task) {}
// recordingEvictor is a delivery.WebhookEvictor that records
// the webhook ids it was asked to evict, so a test can prove
// that a deletion path reached the delivery engine.
type recordingEvictor struct {
mu sync.Mutex
evicted []string
}
func (r *recordingEvictor) EvictWebhook(webhookID string) {
r.mu.Lock()
defer r.mu.Unlock()
r.evicted = append(r.evicted, webhookID)
}
// Evicted returns a copy of the recorded webhook ids.
func (r *recordingEvictor) Evicted() []string {
r.mu.Lock()
defer r.mu.Unlock()
out := make([]string, len(r.evicted))
copy(out, r.evicted)
return out
}
func newTestApp( func newTestApp(
t *testing.T, t *testing.T,
targets ...any, targets ...any,
@@ -74,12 +47,6 @@ func newTestApp(
func() delivery.Notifier { func() delivery.Notifier {
return &noopNotifier{} return &noopNotifier{}
}, },
func() *recordingEvictor {
return &recordingEvictor{}
},
func(r *recordingEvictor) delivery.WebhookEvictor {
return r
},
handlers.New, handlers.New,
), ),
fx.Populate(targets...), fx.Populate(targets...),

View File

@@ -1,356 +0,0 @@
package handlers_test
import (
"context"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"testing"
"github.com/go-chi/chi"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"gorm.io/gorm/clause"
"sneak.berlin/go/webhooker/internal/database"
"sneak.berlin/go/webhooker/internal/handlers"
"sneak.berlin/go/webhooker/internal/session"
)
const (
deleteTestUserID = "test-user-id"
deleteTestUsername = "testuser"
// paramSourceID and paramTargetID are the chi URL parameter
// names the deletion handlers read.
paramSourceID = "sourceID"
paramTargetID = "targetID"
)
// seedWebhook inserts a webhook owned by the test user and
// returns it.
func seedWebhook(
t *testing.T,
db *database.Database,
) *database.Webhook {
t.Helper()
wh := &database.Webhook{
UserID: deleteTestUserID,
Name: "delete-me",
}
require.NoError(
t,
db.DB().Omit(clause.Associations).Create(wh).Error,
)
return wh
}
// seedTarget inserts a target of the given type for a webhook
// and returns it.
func seedTarget(
t *testing.T,
db *database.Database,
webhookID string,
targetType database.TargetType,
) *database.Target {
t.Helper()
tgt := &database.Target{
WebhookID: webhookID,
Name: "t-" + string(targetType),
Type: targetType,
Active: true,
}
require.NoError(
t,
db.DB().Omit(clause.Associations).Create(tgt).Error,
)
return tgt
}
// archivePathFor returns the archive database path the
// delivery engine would use for a webhook: beside the webhook's
// event database in the data directory.
func archivePathFor(
t *testing.T,
mgr *database.WebhookDBManager,
webhookID string,
) string {
t.Helper()
return filepath.Join(
filepath.Dir(mgr.DBPath(webhookID)),
"archive-"+webhookID+".db",
)
}
// writeArchivePlaceholder creates a stand-in archive file so a
// test can assert the file survives webhook deletion.
func writeArchivePlaceholder(path string) error {
return os.WriteFile(path, []byte("archive"), 0o600)
}
// postRequest builds an authenticated POST request carrying the
// given chi URL parameters.
func postRequest(
path string,
cookies []*http.Cookie,
params map[string]string,
) *http.Request {
req := httptest.NewRequestWithContext(
context.Background(), http.MethodPost, path, nil,
)
for _, c := range cookies {
req.AddCookie(c)
}
rctx := chi.NewRouteContext()
for k, v := range params {
rctx.URLParams.Add(k, v)
}
return req.WithContext(
context.WithValue(req.Context(), chi.RouteCtxKey, rctx),
)
}
// TestHandleSourceDelete_EvictsArchiveWriter proves that
// deleting a webhook reaches the delivery engine and releases
// the webhook's archive writer, exercised through the real
// deletion handler rather than by calling the evictor directly.
func TestHandleSourceDelete_EvictsArchiveWriter(t *testing.T) {
t.Parallel()
var (
h *handlers.Handlers
sess *session.Session
db *database.Database
ev *recordingEvictor
)
app := newTestApp(t, &h, &sess, &db, &ev)
app.RequireStart()
t.Cleanup(app.RequireStop)
wh := seedWebhook(t, db)
seedTarget(t, db, wh.ID, database.TargetTypeDatabase)
cookies := authenticatedCookies(
t, sess, deleteTestUserID, deleteTestUsername,
)
req := postRequest(
"/source/"+wh.ID+"/delete",
cookies,
map[string]string{paramSourceID: wh.ID},
)
w := httptest.NewRecorder()
h.HandleSourceDelete().ServeHTTP(w, req)
require.Equal(t, http.StatusSeeOther, w.Code)
assert.Equal(
t, []string{wh.ID}, ev.Evicted(),
"deleting a webhook should evict its archive writer",
)
}
// TestHandleSourceDelete_KeepsArchiveFile proves that deleting
// a webhook does not remove its archive database file: the
// archive is long-term storage the operator owns.
func TestHandleSourceDelete_KeepsArchiveFile(t *testing.T) {
t.Parallel()
var (
h *handlers.Handlers
sess *session.Session
db *database.Database
mgr *database.WebhookDBManager
)
app := newTestApp(t, &h, &sess, &db, &mgr)
app.RequireStart()
t.Cleanup(app.RequireStop)
wh := seedWebhook(t, db)
// Place an archive file where the delivery engine would.
archivePath := archivePathFor(t, mgr, wh.ID)
require.NoError(
t,
writeArchivePlaceholder(archivePath),
)
cookies := authenticatedCookies(
t, sess, deleteTestUserID, deleteTestUsername,
)
req := postRequest(
"/source/"+wh.ID+"/delete",
cookies,
map[string]string{paramSourceID: wh.ID},
)
w := httptest.NewRecorder()
h.HandleSourceDelete().ServeHTTP(w, req)
require.Equal(t, http.StatusSeeOther, w.Code)
assert.FileExists(
t, archivePath,
"webhook deletion must not destroy the archive file",
)
}
// TestHandleTargetDelete_EvictsWhenLastDatabaseTargetGone
// proves that removing the last database target releases the
// archive writer.
func TestHandleTargetDelete_EvictsWhenLastDatabaseTargetGone(
t *testing.T,
) {
t.Parallel()
var (
h *handlers.Handlers
sess *session.Session
db *database.Database
ev *recordingEvictor
)
app := newTestApp(t, &h, &sess, &db, &ev)
app.RequireStart()
t.Cleanup(app.RequireStop)
wh := seedWebhook(t, db)
tgt := seedTarget(
t, db, wh.ID, database.TargetTypeDatabase,
)
cookies := authenticatedCookies(
t, sess, deleteTestUserID, deleteTestUsername,
)
req := postRequest(
"/source/"+wh.ID+"/targets/"+tgt.ID+"/delete",
cookies,
map[string]string{
paramSourceID: wh.ID,
paramTargetID: tgt.ID,
},
)
w := httptest.NewRecorder()
h.HandleTargetDelete().ServeHTTP(w, req)
require.Equal(t, http.StatusSeeOther, w.Code)
assert.Equal(
t, []string{wh.ID}, ev.Evicted(),
"removing the last database target should evict",
)
}
// TestHandleTargetDelete_KeepsWriterWhenDatabaseTargetRemains
// proves that deleting one of several database targets leaves
// the still-needed archive writer alone: the surviving target
// keeps archiving to the same file, so the writer must stay.
func TestHandleTargetDelete_KeepsWriterWhenDatabaseTargetRemains(
t *testing.T,
) {
t.Parallel()
var (
h *handlers.Handlers
sess *session.Session
db *database.Database
ev *recordingEvictor
)
app := newTestApp(t, &h, &sess, &db, &ev)
app.RequireStart()
t.Cleanup(app.RequireStop)
wh := seedWebhook(t, db)
doomed := seedTarget(
t, db, wh.ID, database.TargetTypeDatabase,
)
seedTarget(t, db, wh.ID, database.TargetTypeDatabase)
cookies := authenticatedCookies(
t, sess, deleteTestUserID, deleteTestUsername,
)
req := postRequest(
"/source/"+wh.ID+"/targets/"+doomed.ID+"/delete",
cookies,
map[string]string{
paramSourceID: wh.ID,
paramTargetID: doomed.ID,
},
)
w := httptest.NewRecorder()
h.HandleTargetDelete().ServeHTTP(w, req)
require.Equal(t, http.StatusSeeOther, w.Code)
assert.Empty(
t, ev.Evicted(),
"a second database target still needs the writer",
)
}
// TestHandleTargetDelete_KeepsWriterWhenOtherTypeDeleted proves
// that deleting a target of an unrelated type leaves a
// still-needed archive writer alone: the webhook's database
// target is untouched, so its writer must stay.
func TestHandleTargetDelete_KeepsWriterWhenOtherTypeDeleted(
t *testing.T,
) {
t.Parallel()
var (
h *handlers.Handlers
sess *session.Session
db *database.Database
ev *recordingEvictor
)
app := newTestApp(t, &h, &sess, &db, &ev)
app.RequireStart()
t.Cleanup(app.RequireStop)
wh := seedWebhook(t, db)
seedTarget(t, db, wh.ID, database.TargetTypeDatabase)
other := seedTarget(t, db, wh.ID, database.TargetTypeLog)
cookies := authenticatedCookies(
t, sess, deleteTestUserID, deleteTestUsername,
)
req := postRequest(
"/source/"+wh.ID+"/targets/"+other.ID+"/delete",
cookies,
map[string]string{
paramSourceID: wh.ID,
paramTargetID: other.ID,
},
)
w := httptest.NewRecorder()
h.HandleTargetDelete().ServeHTTP(w, req)
require.Equal(t, http.StatusSeeOther, w.Code)
assert.Empty(
t, ev.Evicted(),
"a surviving database target must keep its writer",
)
}

View File

@@ -533,13 +533,6 @@ func (h *Handlers) deleteWebhookResources(
return return
} }
// Release the delivery engine's per-webhook archiving state
// so a deleted webhook's archive writer (and any handle open
// within its debounce window) does not linger for the
// process lifetime. The archive file itself is deliberately
// left on disk; see evictArchiveWriter.
h.evictArchiveWriter(webhook.ID)
err = h.dbMgr.DeleteDB(webhook.ID) err = h.dbMgr.DeleteDB(webhook.ID)
if err != nil { if err != nil {
h.log.Error( h.log.Error(
@@ -558,64 +551,6 @@ func (h *Handlers) deleteWebhookResources(
http.Redirect(w, r, "/sources", http.StatusSeeOther) http.Redirect(w, r, "/sources", http.StatusSeeOther)
} }
// evictArchiveWriter asks the delivery engine to drop its
// cached archive writer for a webhook, closing the archive file
// handle.
//
// The archive database file is NOT deleted. Unlike the event
// database — which is per-webhook working storage and is
// hard-deleted with the webhook — an archive is explicitly
// long-term storage that an operator may want to keep or move
// away for offline retention. Destroying it as a side effect of
// deleting a webhook would be a surprising and unrecoverable
// data loss, so the file is left for the operator to handle.
func (h *Handlers) evictArchiveWriter(webhookID string) {
if h.evictor == nil {
return
}
h.evictor.EvictWebhook(webhookID)
}
// evictArchiveWriterIfUnused releases a webhook's archive
// writer once the webhook has no database target left to feed
// it.
//
// It is called after any child resource of a webhook is
// deleted, and is correct without knowing which kind was: it
// evicts only when no database target remains, so deleting one
// of several database targets — or deleting an unrelated
// target type — leaves a still-needed writer alone. When no
// database target ever existed there is no writer and eviction
// is a no-op. Soft-deleted targets are excluded by GORM's
// default scope, so the row just deleted is not counted.
func (h *Handlers) evictArchiveWriterIfUnused(webhookID string) {
var remaining int64
err := h.db.DB().
Model(&database.Target{}).
Where(
"webhook_id = ? AND type = ?",
webhookID, database.TargetTypeDatabase,
).
Count(&remaining).Error
if err != nil {
h.log.Error(
"failed to count remaining database targets",
"webhook_id", webhookID,
"error", err,
)
return
}
if remaining > 0 {
return
}
h.evictArchiveWriter(webhookID)
}
// HandleSourceLogs shows the request/response logs for a // HandleSourceLogs shows the request/response logs for a
// webhook. // webhook.
func (h *Handlers) HandleSourceLogs() http.HandlerFunc { func (h *Handlers) HandleSourceLogs() http.HandlerFunc {
@@ -1089,31 +1024,23 @@ func (h *Handlers) HandleEntrypointDelete() http.HandlerFunc {
return h.deleteChildResource( return h.deleteChildResource(
"entrypointID", &database.Entrypoint{}, "entrypointID", &database.Entrypoint{},
"failed to delete entrypoint", "failed to delete entrypoint",
nil,
) )
} }
// HandleTargetDelete handles deleting a target. Deleting the // HandleTargetDelete handles deleting a target.
// last database target of a webhook leaves its archive writer
// with nothing to write, so the writer is evicted and its
// handle closed; the archive file is left on disk.
func (h *Handlers) HandleTargetDelete() http.HandlerFunc { func (h *Handlers) HandleTargetDelete() http.HandlerFunc {
return h.deleteChildResource( return h.deleteChildResource(
"targetID", &database.Target{}, "targetID", &database.Target{},
"failed to delete target", "failed to delete target",
h.evictArchiveWriterIfUnused,
) )
} }
// deleteChildResource returns a handler that deletes a child // deleteChildResource returns a handler that deletes a child
// resource (entrypoint or target) belonging to a webhook. The // resource (entrypoint or target) belonging to a webhook.
// optional afterDelete hook runs with the webhook's id once the
// delete has succeeded, before the redirect.
func (h *Handlers) deleteChildResource( func (h *Handlers) deleteChildResource(
idParam string, idParam string,
model any, model any,
errMsg string, errMsg string,
afterDelete func(webhookID string),
) http.HandlerFunc { ) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) { return func(w http.ResponseWriter, r *http.Request) {
userID, ok := h.getUserID(r) userID, ok := h.getUserID(r)
@@ -1153,10 +1080,6 @@ func (h *Handlers) deleteChildResource(
return return
} }
if afterDelete != nil {
afterDelete(webhook.ID)
}
http.Redirect( http.Redirect(
w, r, w, r,
"/source/"+webhook.ID, "/source/"+webhook.ID,