diff --git a/README.md b/README.md index f0660ec..4690f62 100644 --- a/README.md +++ b/README.md @@ -834,9 +834,25 @@ is both the simplest and the only complete rule: `events-3f2a1c9e-....db`. The only other file is `webhooker.lock`, the always-empty [single-instance lock](#single-instance-lock); it holds no state and is not part of the backup set — a copied one is stale and -blocks nothing. No `-wal` or `-shm` files are produced (see below); a -transient `{name}.db-journal` may exist beside a database while a write -is in flight and is not part of the backup set either. +blocks nothing. + +**`-wal` and `-shm` sidecars.** Every database runs in WAL journal mode, +so while the service is running each `{name}.db` has a `{name}.db-wal` +and a `{name}.db-shm` beside it. **`-wal` is part of the database, not a +scratch file**: it holds committed transactions that are not yet in the +`.db`, so a copy of the `.db` without its `-wal` is missing data and may +have no readable schema at all. `-shm` is regenerable, but there is no +reason to separate the two — copy the directory and you have them. + +A clean shutdown closes `webhooker.db` and every `events-*.db`, which +checkpoints and removes their sidecars; a killed or crashed instance +leaves them, and they must be carried with the `.db`. **Archive +databases are different**: their handle is not closed at shutdown, so +`archive-*.db-wal` and `-shm` normally survive a clean stop and the +`-wal` can hold every row the archive has. Measured on a stopped +instance: `archive-….db` 4096 bytes with no table, its `-wal` 157 KB +holding all 8 archived events. Copying `DATA_DIR` in full is what makes +this a non-issue; copying `.db` files out of it by name is not. Configuration is **not** in `DATA_DIR` — it comes from the environment and from a `.env` file read out of the process working directory. Back @@ -844,17 +860,19 @@ that up with your deployment config, separately. ### A hot copy is not safe -No `journal_mode` pragma is ever issued on any database webhooker opens, -so all of them run on SQLite's default rollback journal. There is no -WAL. The main and event databases are also held open for the entire -process lifetime — `WebhookDBManager` caches event database handles and -closes them only on webhook deletion or shutdown — so "it looked idle" -is not a guarantee that nothing was mid-transaction. +Every database webhooker opens runs in WAL journal mode. The main and +event databases are also held open for the entire process lifetime — +`WebhookDBManager` caches event database handles and closes them only on +webhook deletion or shutdown — so "it looked idle" is not a guarantee +that nothing was mid-transaction. That means `cp`, `rsync`, `tar` or a filesystem snapshot taken against a -running instance can capture a database mid-transaction and yield a file -that is corrupt or missing state the journal would have rolled back. Use -one of the two procedures below instead. +running instance can capture a database and its `-wal` at two different +instants and yield a file that is corrupt or missing state. Copying a +`.db` on its own is worse and fails loudly: recently written pages, +including the schema itself on a young database, live in the `-wal`, so +the copy reads back as an empty or table-less database. Use one of the +two procedures below instead. **Stop, copy, start.** The simplest, needs no extra tooling, and the only one that gives a single point in time across every file: @@ -873,8 +891,9 @@ for db in /path/to/data/*.db; do done ``` -`.backup` takes the proper locks and writes a consistent file. Two -caveats. First, the runtime image is `alpine:3.21` with only +`.backup` reads through the WAL and writes a single consistent file with +no sidecars of its own, so the destination is complete as it stands. +Two caveats. First, the runtime image is `alpine:3.21` with only `ca-certificates` added — the `sqlite3` CLI is **not** in it, so run this on the host against the volume path, or from a throwaway container that mounts the volume. Second, each file is captured at its own @@ -882,15 +901,37 @@ instant, so a webhook created or an event delivered between two files being copied lands in one and not the other. If you need the whole set coherent as of a single moment, stop the service. +Note that `sqlite3 .dump` is **not** one of these procedures: it is +an export, it holds a read transaction open for as long as it runs, and +it pins the WAL against checkpointing for that whole time. It is safe to +run — it does not block ingestion — but back up with `.backup` or a +stopped copy. + Archive databases are the one exception the service is built for: the -archive writer closes its handle after each write (debounced to at most -one reopen per second), so an operator can move `archive-{uuid}.db` -away for offline retention while the service runs, and it is recreated -on the next write (see -[Database Architecture](#database-architecture)). That is a +archive writer closes and reopens its handle around writes (debounced +to at most one reopen per second), so an operator can move +`archive-{uuid}.db` away for offline retention while the service runs, +and it is recreated on the next write. See +[Database Architecture](#database-architecture). That is a move-the-file-away workflow, not a substitute for the backup procedures above. +**Move the sidecars with it.** Under WAL that workflow is no longer a +single file, and the common case is the dangerous one. The reopen +happens on the *next* write after the debounce window elapses, so after +the last write of a burst nothing checkpoints: measured, 20 s after ten +events the `archive-….db` was 4096 bytes — a header, no table — with +all ten rows sitting in a 189 KB `-wal`. Copying the `.db` alone at that +moment yields a file that opens with `no such table: archived_events`. +The file becomes self-contained again when the handle closes, which +happens on the next write past the debounce window, when the connection +pool retires the idle connection (about a minute after the last write), +or at the idle archive sweep — measured, the same file was a complete +20 KB `.db` with no sidecars about a minute after its last write. +Shutdown is **not** on that list: the archive handle is not closed when +the service stops. So either move `archive-{uuid}.db` together with any +`-wal`/`-shm` beside it, or wait until there are none. + ### Restore 1. Stop the service. @@ -904,14 +945,22 @@ above. restored without `webhooker.db` are simply orphaned; nothing references their UUIDs. -3. Do not carry `*.db-journal` files into the restore. Backups taken by - either procedure above are self-consistent and do not need one. +3. Carry any `*.db-wal` and `*.db-shm` files that are in the backup. + They are part of the database, and dropping a `-wal` silently + discards every transaction it still holds. An `.backup` set will not + contain any: it writes a single consolidated file per database. A + stop-and-copy set has none for `webhooker.db` or the `events-*.db`, + because a clean stop closes those and checkpoints their sidecars + away — but it will normally have them for `archive-*.db`, whose + handle stays open across shutdown, and those carry the archive's + rows. A copy salvaged from a crashed instance has them for + everything, and needs all of them. 4. **Fix ownership.** The container runs as the non-root `webhooker` user, UID 1000 / GID 1000. Restored files must be owned by (or writable by) that UID, and so must the directory itself — SQLite - creates the rollback journal beside the database, so a writable file - inside a directory it cannot write is not enough: + creates the `-wal` and `-shm` sidecars beside the database, so a + writable file inside a directory it cannot write is not enough: ```bash chown -R 1000:1000 /path/to/data @@ -1614,10 +1663,12 @@ This separation provides: only, or disables cleanup entirely when set to `0` (retain forever). - **Performance** — each webhook's database has its own page cache and its own lock, so concurrent event ingestion across webhooks won't - contend. No write-ahead log is involved: both DSNs are - `file:{path}?cache=shared&mode=rwc` and no `journal_mode` pragma is - ever issued, so every database runs on SQLite's default rollback - journal. + contend. Every database — main, per-webhook, and archive — is opened + through one code path (`internal/database/sqlite_open.go`) in WAL + journal mode, with a 10-second busy timeout, `BEGIN IMMEDIATE` + transactions, and a bounded connection pool. Under WAL a reader never + blocks a writer, so an operator reading a database does not stall + event ingestion into it. The **database target type** builds on this architecture to provide long-term archiving, separate from the per-webhook event database (which diff --git a/internal/database/database.go b/internal/database/database.go index e92b440..f2880e8 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -4,7 +4,6 @@ package database import ( "context" "crypto/rand" - "database/sql" "encoding/base64" "errors" "fmt" @@ -16,7 +15,6 @@ import ( "go.uber.org/fx" "gorm.io/driver/sqlite" "gorm.io/gorm" - _ "modernc.org/sqlite" // Pure Go SQLite driver "sneak.berlin/go/webhooker/internal/banner" "sneak.berlin/go/webhooker/internal/config" "sneak.berlin/go/webhooker/internal/gormlog" @@ -198,13 +196,11 @@ func (d *Database) connectTo(dataDir string) error { // Construct the main application database path inside DATA_DIR. dbPath := filepath.Join(dataDir, MainDBFileName) - dbURL := fmt.Sprintf( - "file:%s?cache=shared&mode=rwc", - dbPath, - ) - // Open the database with the pure Go SQLite driver - sqlDB, err := sql.Open("sqlite", dbURL) + // Opened through OpenSQLite so this handle carries the same WAL + // journaling, busy timeout, immediate-transaction locking, and pool + // bounds as every other database file. See sqlite_open.go. + sqlDB, err := OpenSQLite(dbPath, SQLiteModeCreate) if err != nil { d.log.Error( "failed to open database", diff --git a/internal/database/sqlite_open.go b/internal/database/sqlite_open.go new file mode 100644 index 0000000..9281185 --- /dev/null +++ b/internal/database/sqlite_open.go @@ -0,0 +1,157 @@ +package database + +import ( + "database/sql" + "fmt" + "net/url" + "time" + + _ "modernc.org/sqlite" // Pure Go SQLite driver +) + +// Every SQLite file this service opens — the main database, the +// per-webhook event databases, and the archive databases — is opened +// through OpenSQLite, so the durability settings below are properties +// of the service rather than of one call site. +// +// modernc.org/sqlite installs no busy handler and issues no pragmas of +// its own: it executes only the pragmas named in explicit `_pragma=` +// DSN parameters, and gorm.io/driver/sqlite adds none when it is +// handed an existing *sql.DB. Every setting therefore has to be +// spelled out here or it is simply not in effect. +// SQLite URI open modes. +const ( + // SQLiteModeCreate creates the database file when it is missing. + SQLiteModeCreate = "rwc" + + // SQLiteModeExisting requires the file to exist already. + SQLiteModeExisting = "rw" +) + +const ( + // SQLiteBusyTimeout is how long SQLite retries a lock conflict + // before returning SQLITE_BUSY. + // + // Under WAL a reader never blocks a writer, so the only conflict + // left is writer against writer: this process's delivery workers + // against each other, or against another process holding the write + // lock. Those clear in milliseconds. Ten seconds is far above that + // and still well inside the receiver's request budget, so an + // inbound webhook waits rather than being rejected with a 500. + SQLiteBusyTimeout = 10 * time.Second + + // sqliteMaxOpenConns bounds the connection pool for one database + // file. + // + // The pool needs a bound at all because database/sql cannot detect + // a connection left mid-transaction: modernc.org/sqlite implements + // neither driver.Validator nor driver.SessionResetter, so a + // connection whose COMMIT failed is returned to the pool with its + // transaction still open and handed out again indefinitely. That is + // what turned four `database is locked` errors into 593 + // `cannot start a transaction within a transaction` in + // https://git.eeqj.de/sneak/webhooker/issues/256. + // + // Four is above the one writer SQLite allows at a time, so reads + // still proceed while a write is in flight, and low enough that + // contention is resolved by the busy handler rather than by piling + // up connections against a lock only one of them can hold. + sqliteMaxOpenConns = 4 + + // sqliteMaxIdleConns keeps the pool warm without holding every + // connection open through an idle period. + sqliteMaxIdleConns = 2 + + // sqliteConnMaxLifetime and sqliteConnMaxIdleTime retire pooled + // connections on a schedule. With _txlock=immediate a failed + // COMMIT should no longer be reachable, but these bound the damage + // if one happens anyway: a poisoned connection is closed and + // replaced within the lifetime instead of wedging the file until + // the process restarts. + sqliteConnMaxLifetime = 5 * time.Minute + sqliteConnMaxIdleTime = time.Minute +) + +// SQLiteDSN builds the connection string for one database file. +// +// mode is the SQLite URI open mode: "rwc" to create the file when it +// is missing, "rw" to require that it already exists. +// +// Three settings carry the fix for +// https://git.eeqj.de/sneak/webhooker/issues/256 and none of them is +// optional: +// +// - journal_mode=WAL, so a reader — an operator running +// `sqlite3 .dump` over their own data — takes a snapshot +// instead of blocking every writer behind it. +// +// - busy_timeout, so a writer that does meet a lock waits for it. +// Without one SQLite gives up immediately; nothing above it +// retries. +// +// - _txlock=immediate, so every transaction takes the write lock at +// BEGIN. A deferred transaction acquires it lazily on its first +// write, and that upgrade returns SQLITE_BUSY *without* consulting +// the busy handler, because SQLite cannot block a transaction that +// may already hold a read snapshot. Such a COMMIT then fails while +// the transaction stays open on the connection. A busy timeout +// alone does not prevent this; BEGIN IMMEDIATE does, by putting +// the wait somewhere the handler applies. +// +// Note what is absent: `cache=shared`. Under a shared cache an +// in-process conflict is reported as SQLITE_LOCKED rather than +// SQLITE_BUSY, and the busy handler does not retry SQLITE_LOCKED — so +// leaving it in would have defeated the busy timeout for exactly the +// contention this service generates. Dropping it is part of the fix, +// not housekeeping. +// +// synchronous is deliberately left at SQLite's default of FULL: this +// is a webhook receiver whose one promise is that an event it answered +// 200 for is durable. +// The order of the _pragma parameters is load-bearing. +// modernc.org/sqlite executes them in the order they appear, on every +// new connection, before the connection is handed to the pool. Setting +// journal_mode first means that pragma itself runs with no busy +// handler installed: the pool opens connections lazily, so the moment +// a new one is created is a moment the database is under load, and +// PRAGMA journal_mode takes a lock. It would fail immediately with +// SQLITE_BUSY and fail the query that caused the connection to be +// opened. busy_timeout is therefore set first, so every pragma after +// it — and the whole life of the connection — is covered. +func SQLiteDSN(path, mode string) string { + q := url.Values{} + q.Set("mode", mode) + q.Set("_txlock", "immediate") + q.Add( + "_pragma", + fmt.Sprintf( + "busy_timeout(%d)", + SQLiteBusyTimeout.Milliseconds(), + ), + ) + q.Add("_pragma", "journal_mode(WAL)") + + return "file:" + path + "?" + q.Encode() +} + +// OpenSQLite opens the SQLite file at path with the service's +// durability settings and pool bounds applied. mode is the SQLite URI +// open mode ("rwc" or "rw"). +// +// The handle is returned rather than a *gorm.DB because the callers +// wrap it in gorm themselves with their own logger. +func OpenSQLite(path, mode string) (*sql.DB, error) { + sqlDB, err := sql.Open("sqlite", SQLiteDSN(path, mode)) + if err != nil { + return nil, fmt.Errorf( + "opening sqlite database %s: %w", path, err, + ) + } + + sqlDB.SetMaxOpenConns(sqliteMaxOpenConns) + sqlDB.SetMaxIdleConns(sqliteMaxIdleConns) + sqlDB.SetConnMaxLifetime(sqliteConnMaxLifetime) + sqlDB.SetConnMaxIdleTime(sqliteConnMaxIdleTime) + + return sqlDB, nil +} diff --git a/internal/database/sqlite_open_test.go b/internal/database/sqlite_open_test.go new file mode 100644 index 0000000..448507c --- /dev/null +++ b/internal/database/sqlite_open_test.go @@ -0,0 +1,178 @@ +package database_test + +import ( + "context" + "path/filepath" + "strings" + "testing" + + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "gorm.io/gorm" + "sneak.berlin/go/webhooker/internal/database" +) + +// livePragma reads a pragma off a live handle. Reading the DSN back +// would prove only that the string was built; these tests assert that +// SQLite actually applied it. +func livePragma(t *testing.T, db *gorm.DB, name string) string { + t.Helper() + + var v string + + row := db.Raw("pragma " + name).Row() + require.NoError(t, row.Scan(&v)) + + return v +} + +func TestSQLiteDSNCarriesTheDurabilitySettings(t *testing.T) { + t.Parallel() + + dsn := database.SQLiteDSN( + "/var/lib/webhooker/webhooker.db", + database.SQLiteModeCreate, + ) + + assert.Contains(t, dsn, "journal_mode%28WAL%29") + assert.Contains(t, dsn, "busy_timeout%2810000%29") + assert.Contains(t, dsn, "_txlock=immediate") + assert.Contains(t, dsn, "mode=rwc") + + // busy_timeout must come first. The driver runs these in order on + // every new connection, and PRAGMA journal_mode takes a lock — a + // connection opened while the database is busy would fail on that + // pragma, with no busy handler yet installed to wait it out. + assert.Less( + t, + strings.Index(dsn, "busy_timeout"), + strings.Index(dsn, "journal_mode"), + "busy_timeout must be applied before journal_mode", + ) + + // cache=shared turns an in-process conflict into SQLITE_LOCKED, + // which the busy handler does not retry. It must never come back. + // See https://git.eeqj.de/sneak/webhooker/issues/256. + assert.NotContains(t, strings.ToLower(dsn), "cache=shared") +} + +// TestPerWebhookDBAppliesPragmasOnALiveHandle is the check the issue +// asks for by name: the settings are confirmed by querying the running +// database, not by inspecting the connection string. +func TestPerWebhookDBAppliesPragmasOnALiveHandle(t *testing.T) { + t.Parallel() + + mgr, lc := setupTestWebhookDBManager(t) + ctx := context.Background() + require.NoError(t, lc.Start(ctx)) + + defer func() { require.NoError(t, lc.Stop(ctx)) }() + + webhookID := uuid.New().String() + + db, err := mgr.GetDB(webhookID) + require.NoError(t, err) + + assert.Equal( + t, "wal", + strings.ToLower(livePragma(t, db, "journal_mode")), + ) + assert.Equal( + t, "10000", livePragma(t, db, "busy_timeout"), + ) +} + +func TestMainDBAppliesPragmasOnALiveHandle(t *testing.T) { + t.Parallel() + + ctx := context.Background() + dir := t.TempDir() + + sqlDB, err := database.OpenSQLite( + filepath.Join(dir, database.MainDBFileName), + database.SQLiteModeCreate, + ) + require.NoError(t, err) + + defer func() { require.NoError(t, sqlDB.Close()) }() + + var journal string + + require.NoError(t, sqlDB. + QueryRowContext(ctx, "pragma journal_mode"). + Scan(&journal)) + assert.Equal(t, "wal", strings.ToLower(journal)) + + var busy string + + require.NoError(t, sqlDB. + QueryRowContext(ctx, "pragma busy_timeout"). + Scan(&busy)) + assert.Equal(t, "10000", busy) +} + +// TestConcurrentReaderDoesNotBlockWrites is the unit-scale form of the +// reproduction in +// https://git.eeqj.de/sneak/webhooker/issues/256: an operator's +// long-held read of their own data used to make every concurrent write +// fail. Under WAL the reader takes a snapshot and the writes proceed. +func TestConcurrentReaderDoesNotBlockWrites(t *testing.T) { + t.Parallel() + + mgr, lc := setupTestWebhookDBManager(t) + ctx := context.Background() + require.NoError(t, lc.Start(ctx)) + + defer func() { require.NoError(t, lc.Stop(ctx)) }() + + webhookID := uuid.New().String() + + db, err := mgr.GetDB(webhookID) + require.NoError(t, err) + + // A second handle on the same file, holding a read transaction + // open across every write below — what `sqlite3 .dump` is. + readerSQL, err := database.OpenSQLite( + mgr.DBPath(webhookID), database.SQLiteModeExisting, + ) + require.NoError(t, err) + + defer func() { require.NoError(t, readerSQL.Close()) }() + + readerConn, err := readerSQL.Conn(ctx) + require.NoError(t, err) + + defer func() { require.NoError(t, readerConn.Close()) }() + + _, err = readerConn.ExecContext(ctx, "begin deferred") + require.NoError(t, err) + + _, err = readerConn.ExecContext( + ctx, "select count(*) from events", + ) + require.NoError(t, err) + + for range 25 { + err = db.Transaction(func(tx *gorm.DB) error { + return tx.Create(&database.Event{ + WebhookID: webhookID, + EntrypointID: uuid.New().String(), + Method: "POST", + Body: "{}", + }).Error + }) + require.NoError(t, err) + } + + _, err = readerConn.ExecContext(ctx, "commit") + require.NoError(t, err) + + var count int64 + + require.NoError( + t, + db.Model(&database.Event{}).Count(&count).Error, + ) + assert.Equal(t, int64(25), count) +} diff --git a/internal/database/webhook_db_manager.go b/internal/database/webhook_db_manager.go index a888ad5..81ca427 100644 --- a/internal/database/webhook_db_manager.go +++ b/internal/database/webhook_db_manager.go @@ -2,7 +2,6 @@ package database import ( "context" - "database/sql" "errors" "fmt" "log/slog" @@ -234,12 +233,11 @@ func (m *WebhookDBManager) openDB( webhookID string, ) (*gorm.DB, error) { path := m.dbPath(webhookID) - dbURL := fmt.Sprintf( - "file:%s?cache=shared&mode=rwc", - path, - ) - sqlDB, err := sql.Open("sqlite", dbURL) + // See sqlite_open.go: WAL, a busy timeout, immediate-transaction + // locking, and a bounded pool, all of which this file needs most — + // it is the one every delivery worker writes to concurrently. + sqlDB, err := OpenSQLite(path, SQLiteModeCreate) if err != nil { return nil, fmt.Errorf( "opening webhook database %s: %w", diff --git a/internal/delivery/engine.go b/internal/delivery/engine.go index 81bdc1f..d347a86 100644 --- a/internal/delivery/engine.go +++ b/internal/delivery/engine.go @@ -41,6 +41,31 @@ const ( // sweep runs. retrySweepInterval = 60 * time.Second + // pendingSweepMinAge is how long a delivery must have sat + // untouched at pending before the sweep will look at it. + // + // It is not what keeps the sweep off live work — inflightSet is, + // and it is exact. This bound sets the re-dispatch cadence for a + // delivery that really is stranded: without it, a delivery the + // database will not let the engine settle would be re-sent on + // every 60-second tick. + // + // It is nonetheless set clear of the longest legitimate attempt, + // so that the two guards do not both have to be right. That + // length is MaxTargetTimeoutSeconds (300s), the per-target + // timeout the target form accepts — not httpClientTimeout, which + // is merely the default. Fifteen minutes leaves a margin of + // three times the ceiling rather than the zero margin the two + // equal values would have given. + pendingSweepMinAge = 15 * time.Minute + + // pendingSweepBatch bounds how many stranded pending deliveries + // one sweep of one webhook re-dispatches. The sweep runs every + // retrySweepInterval, so a larger backlog drains across + // successive sweeps instead of arriving as one burst against a + // database that was already struggling to accept writes. + pendingSweepBatch = 500 + // MaxInlineBodySize is the maximum event body size that // will be carried inline in a Task through the channel. // Bodies at or above this size are left nil and fetched @@ -157,6 +182,12 @@ type Engine struct { // dbTarget is retained so the engine can reach the archive // writer registry for webhook eviction and the idle sweep. dbTarget *databaseTarget + + // inflight is the set of deliveries this engine currently owns. + // Recovery and the sweeps re-dispatch only what it does not + // hold. Held by value: its zero value works, so no constructor + // can leave it out. See inflight.go. + inflight inflightSet } // New creates and registers the delivery engine with the @@ -189,12 +220,27 @@ func New( // are ready. func (e *Engine) Notify(tasks []Task) { for i := range tasks { + // Owned before it is queued, and until the worker that runs + // it returns. A task can sit in a 10000-deep channel for a + // long time on a healthy system, and nothing may re-send it + // while it waits. See inflight.go. + if !e.inflight.retainIdle(tasks[i].DeliveryID) { + e.log.Warn( + "delivery already in flight, not queued again", + "delivery_id", tasks[i].DeliveryID, + "event_id", tasks[i].EventID, + ) + + continue + } + select { case e.deliveryCh <- tasks[i]: default: + e.inflight.release(tasks[i].DeliveryID) e.log.Warn( "delivery channel full, "+ - "task will be recovered on restart", + "task will be recovered by the sweep", "delivery_id", tasks[i].DeliveryID, "event_id", tasks[i].EventID, ) @@ -229,10 +275,20 @@ func (e *Engine) ScheduleRetry( "next_attempt", task.AttemptNum, ) + // The reference is taken here rather than when the timer fires, + // so the delivery stays owned across the whole backoff window. + // Its caller is a target inside Deliver, so the engine already + // owns it; this second reference is what keeps that ownership + // alive after the worker returns and the row sits at retrying + // with nothing running. Without it the sweep finds the row + // orphaned and sends it again. + e.inflight.retain(task.DeliveryID) + time.AfterFunc(delay, func() { select { case e.retryCh <- task: default: + e.inflight.release(task.DeliveryID) e.log.Warn( "retry channel full, delivery "+ "will be recovered by periodic sweep", @@ -332,13 +388,35 @@ func (e *Engine) worker(ctx context.Context) { case <-ctx.Done(): return case task := <-e.deliveryCh: - e.processNewTask(ctx, &task) + e.runTask(ctx, &task, e.processNewTask) case task := <-e.retryCh: - e.processRetryTask(ctx, &task) + e.runTask(ctx, &task, e.processRetryTask) } } } +// runTask runs one task and then drops the reference the queueing +// side took on its delivery. +// +// The release is deferred rather than written after the call because +// every early return inside the processing paths must drop it too: a +// delivery whose database could not be opened is one the engine has +// stopped working on, and leaving it owned would hide it from the +// sweep forever. +// +// Ownership does not necessarily end here. A target that scheduled a +// retry took its own reference before this one is dropped, so the +// delivery stays owned through the backoff window. +func (e *Engine) runTask( + ctx context.Context, + task *Task, + run func(context.Context, *Task), +) { + defer e.inflight.release(task.DeliveryID) + + run(ctx, task) +} + func (e *Engine) recoverPending(ctx context.Context) { defer e.wg.Done() @@ -526,7 +604,16 @@ func (e *Engine) recoverRetryingDeliveries( return } + settled := e.reconcileDelivered( + webhookDB, webhookID, retrying, + e.loadTargetMap(retrying), + ) + for i := range retrying { + if _, ok := settled[retrying[i].ID]; ok { + continue + } + e.recoverSingleRetry( webhookDB, webhookID, &retrying[i], ) @@ -588,6 +675,10 @@ func (e *Engine) recoverSingleRetry( d, webhookID, &event, &target, attemptNum+1, ) + if !e.rescheduleRecovered(webhookDB, task, remaining) { + return + } + e.log.Info( "recovering retrying delivery", "webhook_id", webhookID, @@ -595,8 +686,6 @@ func (e *Engine) recoverSingleRetry( "attempt", attemptNum, "remaining_backoff", remaining, ) - - e.ScheduleRetry(task, remaining) } func (e *Engine) recoverPendingDeliveries( @@ -606,12 +695,14 @@ func (e *Engine) recoverPendingDeliveries( ) { var deliveries []database.Delivery + // No Preload: event bodies are read one at a time in + // sendRecoveredDeliveries, and only for the deliveries actually + // being sent. result := webhookDB. Where( "status = ?", database.DeliveryStatusPending, ). - Preload("Event"). Find(&deliveries) if result.Error != nil { @@ -634,11 +725,137 @@ func (e *Engine) recoverPendingDeliveries( "count", len(deliveries), ) + e.recoverPendingBatch( + ctx, webhookDB, webhookID, deliveries, + ) +} + +// recoverPendingBatch settles every delivery in the batch that was +// already delivered, and re-dispatches only the rest. Both the +// restart-time recovery and the periodic sweep go through it, so a +// pending delivery is treated the same however it was found. +func (e *Engine) recoverPendingBatch( + ctx context.Context, + webhookDB *gorm.DB, + webhookID string, + deliveries []database.Delivery, +) { targetMap := e.loadTargetMap(deliveries) - e.sendRecoveredDeliveries( - ctx, deliveries, webhookID, targetMap, + settled := e.reconcileDelivered( + webhookDB, webhookID, deliveries, targetMap, ) + + e.sendRecoveredDeliveries( + ctx, webhookDB, deliveries, webhookID, + targetMap, settled, + ) +} + +// reconcileDelivered finds the deliveries in a recovered batch that +// already have a successful DeliveryResult, marks them delivered, and +// returns their ids so the caller does not send them a second time. +// +// This is the state the engine previously had no way to represent. A +// delivery is left in a non-terminal state by a failed bookkeeping +// write, and that covers two different histories: nothing was ever +// sent, or the send reached the receiver and only the status write +// failed. Re-sending was the sole option, so every stranded row +// produced a duplicate at the receiver and an event log that recorded +// one attempt for two POSTs. A successful result row distinguishes +// them: it is written before the status, so its presence means the +// wire I/O happened and was recorded, and all that is missing is the +// status. +// +// Every recovery path runs this, not only the pending one. A delivery +// abandoned at retrying can hold a successful result just as a pending +// one can — a second attempt that reached the receiver and whose status +// write then failed sits at retrying with success recorded — and +// re-sending it is the same duplicate. +// +// Deliveries whose result row itself never landed are not in the +// returned set and are re-sent, recorded as the further attempt they +// are. That is honest at-least-once delivery rather than a silent +// duplicate. +func (e *Engine) reconcileDelivered( + webhookDB *gorm.DB, + webhookID string, + deliveries []database.Delivery, + targetMap map[string]database.Target, +) map[string]struct{} { + settled := make(map[string]struct{}) + + if len(deliveries) == 0 { + return settled + } + + ids := make([]string, 0, len(deliveries)) + for i := range deliveries { + ids = append(ids, deliveries[i].ID) + } + + var deliveredIDs []string + + err := webhookDB. + Model(&database.DeliveryResult{}). + Where( + "delivery_id IN ? AND success = ?", ids, true, + ). + Distinct(). + Pluck("delivery_id", &deliveredIDs).Error + if err != nil { + // Every delivery stays out of the settled set, so the batch + // is re-sent exactly as it was before this check existed. + // That is the safe direction: a duplicate delivery beats + // declaring a delivery successful on a query that failed. + e.log.Error( + "failed to query successful delivery results; "+ + "pending deliveries will be re-sent", + "webhook_id", webhookID, + "error", err, + ) + + return settled + } + + for _, id := range deliveredIDs { + settled[id] = struct{}{} + } + + if len(settled) == 0 { + return settled + } + + e.log.Info( + "settling recovered deliveries that already succeeded", + "webhook_id", webhookID, + "count", len(settled), + ) + + for i := range deliveries { + if _, ok := settled[deliveries[i].ID]; !ok { + continue + } + + // A delivery the engine is working on right now settles + // itself; writing over it from here would race that worker. + if !e.inflight.retainIdle(deliveries[i].ID) { + delete(settled, deliveries[i].ID) + + continue + } + + e.settleStatus( + webhookDB, + &deliveries[i], + targetMap[deliveries[i].TargetID].Type, + database.DeliveryStatusDelivered, + ) + + e.inflight.release(deliveries[i].ID) + } + + return settled } func (e *Engine) retrySweep(ctx context.Context) { @@ -722,6 +939,11 @@ func (e *Engine) sweepWebhookRetries( return } + settled := e.reconcileDelivered( + webhookDB, webhookID, retrying, + e.loadTargetMap(retrying), + ) + for i := range retrying { select { case <-ctx.Done(): @@ -729,10 +951,70 @@ func (e *Engine) sweepWebhookRetries( default: } + if _, ok := settled[retrying[i].ID]; ok { + continue + } + e.sweepSingleRetry( webhookDB, webhookID, &retrying[i], ) } + + e.sweepWebhookPending(ctx, webhookDB, webhookID) +} + +// sweepWebhookPending recovers deliveries stranded at pending. +// +// A delivery is created pending and leaves that state only when its +// outcome is written, so a pending row the engine does not own is one +// whose bookkeeping write failed — the state that used to sit there +// until a restart, and then produce a duplicate at the receiver. The +// sweep gives it the same reconcile-then-dispatch treatment restart +// recovery gets, so it costs a minute rather than an operator +// noticing. +// +// What keeps the sweep off live work is ownership, checked per +// delivery in takeForRedispatch, not the age bound in this query. +// A delivery waiting in deliveryCh is pending and arbitrarily old — +// the channel holds 10000 tasks and 10 workers drain it — so +// reasoning from the row's age alone re-sends it. See inflight.go. +func (e *Engine) sweepWebhookPending( + ctx context.Context, + webhookDB *gorm.DB, + webhookID string, +) { + var pending []database.Delivery + + err := webhookDB. + Where( + "status = ? AND updated_at < ?", + database.DeliveryStatusPending, + time.Now().Add(-pendingSweepMinAge), + ). + Limit(pendingSweepBatch). + Find(&pending).Error + if err != nil { + e.log.Error( + "retry sweep: "+ + "failed to query pending deliveries", + "webhook_id", webhookID, + "error", err, + ) + + return + } + + if len(pending) == 0 { + return + } + + e.log.Info( + "retry sweep: recovering stranded pending deliveries", + "webhook_id", webhookID, + "count", len(pending), + ) + + e.recoverPendingBatch(ctx, webhookDB, webhookID, pending) } // sweepSingleRetry re-enqueues an orphaned retrying delivery @@ -789,17 +1071,20 @@ func (e *Engine) sweepSingleRetry( d, webhookID, &event, &target, attemptNum+1, ) - select { - case e.retryCh <- task: - e.log.Info( - "retry sweep: "+ - "recovered orphaned retrying delivery", - "delivery_id", d.ID, - "webhook_id", webhookID, - "attempt", attemptNum+1, - ) - default: + if !e.redispatch( + e.retryCh, webhookDB, task, + database.DeliveryStatusRetrying, + ) { + return } + + e.log.Info( + "retry sweep: "+ + "recovered orphaned retrying delivery", + "delivery_id", d.ID, + "webhook_id", webhookID, + "attempt", attemptNum+1, + ) } // failUnretryableRetry terminally fails an orphaned retrying @@ -822,6 +1107,16 @@ func (e *Engine) failUnretryableRetry( d *database.Delivery, target *database.Target, ) { + // Terminal, and reached from the recovery paths, so it takes + // ownership like every other write they make: a delivery the + // engine is still attempting must not be failed underneath the + // worker running it. + if !e.inflight.retainIdle(d.ID) { + return + } + + defer e.inflight.release(d.ID) + e.log.Warn( "failing orphaned retrying delivery: target "+ "type no longer supports retries", @@ -839,7 +1134,7 @@ func (e *Engine) failUnretryableRetry( target.Type, ) - e.recordResult( + err := e.recordResult( webhookDB, d, e.countAttempts(webhookDB, d.ID)+1, @@ -849,6 +1144,11 @@ func (e *Engine) failUnretryableRetry( reason, 0, ) + if err != nil { + e.bookkeepingFailed(d, err) + + return + } // The type is passed rather than assigned onto d: the delivery // is loaded here without its target relation, and populating @@ -856,7 +1156,7 @@ func (e *Engine) failUnretryableRetry( // whole target row — plaintext config, which for a slack target // is the credential — into the per-webhook event database. See // https://git.eeqj.de/sneak/webhooker/issues/206. - e.updateDeliveryStatus( + e.settleStatus( webhookDB, d, target.Type, database.DeliveryStatusFailed, ) @@ -878,7 +1178,7 @@ func (e *Engine) processDelivery( "type", d.Target.Type, ) - e.updateDeliveryStatus( + e.settleStatus( webhookDB, d, d.Target.Type, database.DeliveryStatusFailed, ) @@ -910,6 +1210,14 @@ func (e *Engine) observeAttempt( // recordResult persists a DeliveryResult row describing a // single attempt. It is a cross-target helper the targets // call. +// +// It returns its error rather than swallowing it. A DeliveryResult +// row is the only record that an attempt happened at all, so a +// caller that ignored a failed write would go on to mark the +// delivery delivered — leaving the event log claiming one attempt +// for a receiver that got two. Every caller must instead stop +// advancing the delivery's status and let it stay in the +// non-terminal state it already holds; see bookkeepingFailed. func (e *Engine) recordResult( webhookDB *gorm.DB, d *database.Delivery, @@ -918,7 +1226,7 @@ func (e *Engine) recordResult( statusCode int, respBody, errMsg string, durationMs int64, -) { +) error { result := &database.DeliveryResult{ DeliveryID: d.ID, AttemptNum: attemptNum, @@ -931,12 +1239,44 @@ func (e *Engine) recordResult( err := webhookDB.Create(result).Error if err != nil { - e.log.Error( - "failed to record delivery result", - "delivery_id", d.ID, - "error", err, + return fmt.Errorf( + "recording delivery result for %s: %w", d.ID, err, ) } + + return nil +} + +// bookkeepingFailed reports that a delivery's own record of what +// happened could not be written, and deliberately writes nothing in +// response. +// +// Leaving the row alone is the whole point. A delivery is created +// pending and only ever leaves that state through +// updateDeliveryStatus, so a delivery whose bookkeeping write failed +// is still pending or retrying — the two non-terminal states, per +// DeliveryStatus.Terminal — and both are swept and recovered. Writing +// anything here would need the very database that just refused a +// write, and would be one more thing to fail; not writing cannot. +// +// The cost is honest at-least-once behaviour: a send that reached the +// receiver but whose result row did not land is attempted again, and +// recorded as the further attempt it is. What no longer happens is the +// silent duplicate — a second POST the event log denies ever +// occurred. Where the result row *did* land and only the status write +// failed, reconcileDelivered settles the row without re-sending. +func (e *Engine) bookkeepingFailed( + d *database.Delivery, err error, +) { + e.log.Error( + "delivery bookkeeping write failed; leaving delivery "+ + "in a recoverable state", + "delivery_id", d.ID, + "event_id", d.EventID, + "target_id", d.TargetID, + "status", d.Status, + "error", err, + ) } // updateDeliveryStatus persists a new status for a delivery. @@ -952,26 +1292,51 @@ func (e *Engine) recordResult( // // The counter moves only after the row is written, so a transition // the database rejected is not claimed as an outcome that happened. +// For the same reason the error is returned rather than logged and +// dropped: a delivery whose status write failed has not reached that +// status, and its caller must not act as though it had. func (e *Engine) updateDeliveryStatus( webhookDB *gorm.DB, d *database.Delivery, targetType database.TargetType, status database.DeliveryStatus, -) { +) error { err := webhookDB.Model(d). Update("status", status).Error if err != nil { - e.log.Error( - "failed to update delivery status", - "delivery_id", d.ID, - "status", status, - "error", err, + return fmt.Errorf( + "updating delivery %s to status %s: %w", + d.ID, status, err, ) - - return } - e.mtr.DeliveryStatusChanged(targetType, status) + // An empty type means the target row is gone — a delivery being + // settled long after its target was deleted. The row still has to + // be settled, but the counter is left alone rather than given a + // series labelled with the empty string. + if targetType != "" { + e.mtr.DeliveryStatusChanged(targetType, status) + } + + return nil +} + +// settleStatus moves a delivery to its outcome status and reports a +// failed write through bookkeepingFailed, which leaves the row +// recoverable. It exists so the target call sites read as one +// statement rather than four lines of identical error handling. +func (e *Engine) settleStatus( + webhookDB *gorm.DB, + d *database.Delivery, + targetType database.TargetType, + status database.DeliveryStatus, +) { + err := e.updateDeliveryStatus( + webhookDB, d, targetType, status, + ) + if err != nil { + e.bookkeepingFailed(d, err) + } } func truncate(s string, maxLen int) string { @@ -1065,6 +1430,184 @@ func (e *Engine) countAttempts( return int(resultCount) } +// takeForRedispatch decides whether a recovered delivery may be sent +// again, and takes it if so. It is the single gate every re-dispatch +// path goes through, and it asks two separate questions in order. +// +// First, does the engine already own this delivery? Ownership is +// exact and mutually exclusive, so a delivery queued, being attempted, +// or waiting out a retry backoff is refused here, and two dispatchers +// racing for the same delivery cannot both win. See inflight.go. +// +// Second, is the row still in the status that made it eligible? The +// batch was read some time ago and a worker may have settled a row +// since. The check is a conditional update rather than a read so the +// answer cannot go stale between asking and acting. +// +// Stamping updated_at is the same statement, and it is a cadence +// control rather than a claim: the pending sweep selects on that +// column, so a delivery handed out now is not selected again on the +// next tick a minute later but after pendingSweepMinAge. A delivery +// the database refuses to settle is therefore retried on that +// interval instead of every tick. +// +// A failed write is a refusal. It means the database is not accepting +// writes, which is the condition that stranded the delivery in the +// first place; an attempt that cannot be recorded is exactly the +// unlogged duplicate this is all here to prevent. +// +// The caller must release ownership if it then fails to queue the +// task. +func (e *Engine) takeForRedispatch( + webhookDB *gorm.DB, + deliveryID string, + eligible database.DeliveryStatus, +) bool { + if !e.inflight.retainIdle(deliveryID) { + return false + } + + res := webhookDB. + Model(&database.Delivery{}). + Where( + "id = ? AND status = ?", deliveryID, eligible, + ). + UpdateColumn("updated_at", time.Now()) + + if res.Error != nil { + e.log.Error( + "failed to mark delivery for re-dispatch; "+ + "leaving it for a later sweep", + "delivery_id", deliveryID, + "error", res.Error, + ) + e.inflight.release(deliveryID) + + return false + } + + if res.RowsAffected != 1 { + // Settled underneath us between the query and here. + e.inflight.release(deliveryID) + + return false + } + + return true +} + +// queueRecovered puts an owned delivery's task on a worker channel, +// dropping the ownership the gate took if it does not fit. +func (e *Engine) queueRecovered( + ch chan<- Task, task Task, +) bool { + select { + case ch <- task: + return true + default: + e.inflight.release(task.DeliveryID) + e.log.Warn( + "worker channel full during recovery; "+ + "delivery will be recovered by a later sweep", + "delivery_id", task.DeliveryID, + "webhook_id", task.WebhookID, + ) + + return false + } +} + +// redispatch hands a recovered delivery to a worker channel through +// takeForRedispatch, and reports whether the task was queued. +func (e *Engine) redispatch( + ch chan<- Task, + webhookDB *gorm.DB, + task Task, + eligible database.DeliveryStatus, +) bool { + if !e.takeForRedispatch( + webhookDB, task.DeliveryID, eligible, + ) { + return false + } + + return e.queueRecovered(ch, task) +} + +// rescheduleRecovered hands an orphaned retrying delivery back to the +// retry timer, through the same gate. It reports whether the delivery +// was rescheduled. +// +// The reference taken by the gate is dropped as soon as ScheduleRetry +// has taken its own, which it does before returning: what keeps the +// delivery owned through the backoff window is ScheduleRetry's +// reference, not this one. +func (e *Engine) rescheduleRecovered( + webhookDB *gorm.DB, task Task, delay time.Duration, +) bool { + if !e.takeForRedispatch( + webhookDB, task.DeliveryID, + database.DeliveryStatusRetrying, + ) { + return false + } + + defer e.inflight.release(task.DeliveryID) + + e.ScheduleRetry(task, delay) + + return true +} + +// countAttemptsBatch counts the recorded attempts of every delivery +// in a batch with one grouped query, keyed by delivery id. Deliveries +// with no attempts are simply absent from the result, which reads back +// as the zero this caller wants. +// +// One query rather than one per delivery: this runs on the recovery +// path, which is a burst of writes against a database that has just +// been under enough contention to strand these rows in the first +// place. See https://git.eeqj.de/sneak/webhooker/issues/256. +func (e *Engine) countAttemptsBatch( + webhookDB *gorm.DB, deliveries []database.Delivery, +) map[string]int { + counts := make(map[string]int, len(deliveries)) + + if len(deliveries) == 0 { + return counts + } + + ids := make([]string, 0, len(deliveries)) + for i := range deliveries { + ids = append(ids, deliveries[i].ID) + } + + // One delivery id per recorded attempt, tallied here rather than + // grouped in SQL: internal/gormlog forbids (*gorm.DB).Scan, which + // a GROUP BY into a struct would need, and an attempt row per + // delivery is bounded by the target's MaxRetries. + var attemptIDs []string + + err := webhookDB. + Model(&database.DeliveryResult{}). + Where("delivery_id IN ?", ids). + Pluck("delivery_id", &attemptIDs).Error + if err != nil { + e.log.Error( + "failed to count delivery attempts for recovery", + "error", err, + ) + + return counts + } + + for _, id := range attemptIDs { + counts[id]++ + } + + return counts +} + func (e *Engine) loadEvent( webhookDB *gorm.DB, eventID string, ) (database.Event, error) { @@ -1132,6 +1675,10 @@ func buildRecoveryTask( func (e *Engine) loadTargetMap( deliveries []database.Delivery, ) map[string]database.Target { + if len(deliveries) == 0 { + return nil + } + seen := make(map[string]bool) targetIDs := make([]string, 0, len(deliveries)) @@ -1168,12 +1715,33 @@ func (e *Engine) loadTargetMap( return targetMap } +// sendRecoveredDeliveries re-dispatches pending deliveries, skipping +// the ids in settled — those already reached their receiver and have +// been marked delivered by reconcileDelivered. +// +// The skip and takeForRedispatch's status check answer different +// questions and neither replaces the other. This one is "did this +// delivery already succeed", which is what settles the row to +// delivered instead of sending it, and which is the only thing that +// keeps the retrying paths from terminally failing a delivery that +// reconcile just settled. The status check is "is the row still what +// the batch query said it was", which catches a worker settling it to +// anything at all in between. func (e *Engine) sendRecoveredDeliveries( ctx context.Context, + webhookDB *gorm.DB, deliveries []database.Delivery, webhookID string, targetMap map[string]database.Target, + settled map[string]struct{}, ) { + // The attempt number continues each delivery's own history + // rather than restarting at 1. A recovered delivery may already + // have recorded attempts, and numbering the next one 1 again + // both collides in the event log and hands the retry path a + // backoff computed from the wrong attempt. + attempts := e.countAttemptsBatch(webhookDB, deliveries) + for i := range deliveries { select { case <-ctx.Done(): @@ -1181,6 +1749,10 @@ func (e *Engine) sendRecoveredDeliveries( default: } + if _, ok := settled[deliveries[i].ID]; ok { + continue + } + target, ok := targetMap[deliveries[i].TargetID] if !ok { e.log.Error( @@ -1192,22 +1764,40 @@ func (e *Engine) sendRecoveredDeliveries( continue } + if !e.takeForRedispatch( + webhookDB, deliveries[i].ID, + database.DeliveryStatusPending, + ) { + continue + } + + // The body is read here, one delivery at a time and only for + // deliveries that are actually being sent, rather than + // preloaded across the whole batch. A batch is up to + // pendingSweepBatch rows at up to the 1 MB ingest cap, and + // most of a sweep's batch is refused by the gate above — so + // preloading would hold hundreds of megabytes per webhook per + // tick to build tasks it then discards. + event, err := e.loadEvent( + webhookDB, deliveries[i].EventID, + ) + if err != nil { + e.log.Error( + "failed to load event for recovered delivery", + "delivery_id", deliveries[i].ID, + "event_id", deliveries[i].EventID, + "error", err, + ) + e.inflight.release(deliveries[i].ID) + + continue + } + task := buildRecoveryTask( - &deliveries[i], webhookID, - &deliveries[i].Event, &target, 1, + &deliveries[i], webhookID, &event, &target, + attempts[deliveries[i].ID]+1, ) - select { - case e.deliveryCh <- task: - default: - e.log.Warn( - "delivery channel full during "+ - "recovery, remaining deliveries "+ - "will be recovered on next restart", - "delivery_id", deliveries[i].ID, - ) - - return - } + e.queueRecovered(e.deliveryCh, task) } } diff --git a/internal/delivery/engine_integration_test.go b/internal/delivery/engine_integration_test.go index 2976a2b..ba7fdc1 100644 --- a/internal/delivery/engine_integration_test.go +++ b/internal/delivery/engine_integration_test.go @@ -2,7 +2,6 @@ package delivery_test import ( "context" - "database/sql" "encoding/json" "fmt" "io" @@ -70,11 +69,12 @@ func iMainDB(t *testing.T) *gorm.DB { t.TempDir(), "main-test.db", ) - dsn := fmt.Sprintf( - "file:%s?cache=shared&mode=rwc", dbPath, + // Opened the way the service opens the main database, so these + // tests cannot pass against journal and locking settings + // production does not use. + sqlDB, err := database.OpenSQLite( + dbPath, database.SQLiteModeCreate, ) - - sqlDB, err := sql.Open("sqlite", dsn) require.NoError(t, err) t.Cleanup(func() { _ = sqlDB.Close() }) diff --git a/internal/delivery/engine_test.go b/internal/delivery/engine_test.go index 2376de7..6c2e0a2 100644 --- a/internal/delivery/engine_test.go +++ b/internal/delivery/engine_test.go @@ -3,7 +3,6 @@ package delivery_test import ( "bytes" "context" - "database/sql" "encoding/json" "fmt" "log/slog" @@ -37,11 +36,12 @@ func testWebhookDB(t *testing.T) *gorm.DB { t.TempDir(), "events-test.db", ) - dsn := fmt.Sprintf( - "file:%s?cache=shared&mode=rwc", dbPath, + // Opened the way the service opens a per-webhook database, so + // these tests cannot pass against journal and locking settings + // production does not use. + sqlDB, err := database.OpenSQLite( + dbPath, database.SQLiteModeCreate, ) - - sqlDB, err := sql.Open("sqlite", dsn) require.NoError(t, err) t.Cleanup(func() { _ = sqlDB.Close() }) diff --git a/internal/delivery/export_test.go b/internal/delivery/export_test.go index 97763f4..e3b20c2 100644 --- a/internal/delivery/export_test.go +++ b/internal/delivery/export_test.go @@ -33,6 +33,11 @@ const ( // response is written against this number, so a test has to // be able to name it. ExportMaxBodyLog = maxBodyLog + + // ExportPendingSweepMinAge is how long a delivery must sit at + // pending before the sweep treats it as stranded. A test has to + // name it to age a row past the bound. + ExportPendingSweepMinAge = pendingSweepMinAge ) // ExportIsBlockedIP exposes isBlockedIP for testing. @@ -286,6 +291,26 @@ func (e *Engine) ExportWedgeWorker(release <-chan struct{}) { }) } +// ExportInflightHeld reports how many deliveries the engine currently +// owns, so a test can prove ownership is released rather than leaked. +func (e *Engine) ExportInflightHeld() int { + return e.inflight.held() +} + +// ExportRetainDelivery takes the first reference on a delivery, as the +// queueing side does. It lets a test put a delivery into the state a +// worker or a full channel would, without running the pool. +func (e *Engine) ExportRetainDelivery(deliveryID string) bool { + return e.inflight.retainIdle(deliveryID) +} + +// ExportRecoverRetryingDeliveries exposes recoverRetryingDeliveries. +func (e *Engine) ExportRecoverRetryingDeliveries( + webhookDB *gorm.DB, webhookID string, +) { + e.recoverRetryingDeliveries(webhookDB, webhookID) +} + // ExportDeliveryCh returns the delivery channel. func (e *Engine) ExportDeliveryCh() chan Task { return e.deliveryCh diff --git a/internal/delivery/inflight.go b/internal/delivery/inflight.go new file mode 100644 index 0000000..9596e30 --- /dev/null +++ b/internal/delivery/inflight.go @@ -0,0 +1,110 @@ +package delivery + +import "sync" + +// inflightSet records which deliveries the engine currently owns. +// +// A delivery is owned from the moment a task for it is handed to a +// channel or to a retry timer until the engine has no further plan for +// it in memory. Restart recovery and both arms of the periodic sweep +// re-dispatch only deliveries the set does not hold, which is what +// makes them exact rather than a guess about how long a row has sat at +// pending. +// +// This replaces reasoning from timestamps. A delivery's row says +// pending from creation until its outcome is written, which covers +// four different situations — never dispatched, waiting in a channel, +// being attempted right now, and genuinely stranded — and no column +// distinguishes them. Only the engine knows which, and it knows +// exactly. `deliveryChannelSize` is 10000 against 10 workers, so a +// perfectly healthy delivery can wait far longer than any age bound +// worth setting before its attempt even begins; an age bound alone +// re-sends it. See +// https://git.eeqj.de/sneak/webhooker/issues/256. +// +// In-memory state is sufficient because a data directory admits one +// process: internal/datadir takes an flock on it at startup and a +// second instance refuses to run. Deliveries owned by a process that +// died are not in any successor's set, and restart recovery is what +// picks those up. +// +// References are counted rather than held as a plain set because +// ownership outlives the worker that took it. A target that schedules +// a retry from inside Deliver adds a reference while the worker still +// holds one, so the delivery stays owned across the gap between the +// worker returning and the timer firing — the window in which a sweep +// would otherwise find the row at retrying and send it again. +// +// The zero value is ready to use, and the Engine holds one by value. +// That is deliberate: an engine built by a constructor that forgot to +// initialise this would not refuse to re-dispatch anything, and the +// symptom would be duplicate deliveries rather than a failure anybody +// notices. +type inflightSet struct { + mu sync.Mutex + ids map[string]int +} + +// retain adds a reference to a delivery the caller already knows the +// engine owns, so that ownership survives the current holder letting +// go. It cannot fail. +func (s *inflightSet) retain(deliveryID string) { + s.mu.Lock() + defer s.mu.Unlock() + + if s.ids == nil { + s.ids = make(map[string]int) + } + + s.ids[deliveryID]++ +} + +// retainIdle takes the first reference on a delivery, and reports +// whether it got it. It fails when the engine already owns the +// delivery, which is what makes two claimants — restart recovery and +// the sweep run concurrently, or two sweep arms — mutually exclusive +// rather than merely atomic. +func (s *inflightSet) retainIdle(deliveryID string) bool { + s.mu.Lock() + defer s.mu.Unlock() + + if s.ids[deliveryID] > 0 { + return false + } + + if s.ids == nil { + s.ids = make(map[string]int) + } + + s.ids[deliveryID] = 1 + + return true +} + +// release drops one reference. The delivery becomes eligible for +// re-dispatch again once the last one goes. +func (s *inflightSet) release(deliveryID string) { + s.mu.Lock() + defer s.mu.Unlock() + + n := s.ids[deliveryID] - 1 + if n <= 0 { + delete(s.ids, deliveryID) + + return + } + + s.ids[deliveryID] = n +} + +// held reports how many deliveries the engine currently owns. It +// exists so a test can assert that ownership is released rather than +// leaked: a reference that is never dropped hides its delivery from +// every sweep for the life of the process, which is the one way this +// mechanism can fail silently. +func (s *inflightSet) held() int { + s.mu.Lock() + defer s.mu.Unlock() + + return len(s.ids) +} diff --git a/internal/delivery/inflight_test.go b/internal/delivery/inflight_test.go new file mode 100644 index 0000000..1c86003 --- /dev/null +++ b/internal/delivery/inflight_test.go @@ -0,0 +1,428 @@ +package delivery_test + +import ( + "context" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "sneak.berlin/go/webhooker/internal/database" + "sneak.berlin/go/webhooker/internal/delivery" +) + +// These tests pin the rule that decides whether a delivery may be +// handed back to a worker: the engine re-dispatches only what it does +// not already own. Age alone is not that rule — a healthy delivery +// waiting in a 10000-deep channel is old and must not be re-sent. See +// https://git.eeqj.de/sneak/webhooker/issues/256. + +// fSweepSetup seeds the main database with the webhook row the sweep +// enumerates, and returns the setup. +func fSweepSetup( + t *testing.T, targetID, name string, +) iSetup { + t.Helper() + + s := newISetup(t) + + iCreateTarget(t, s.MainDB, targetID, + s.WebhookID, name, + database.TargetTypeLog, "", 0, + ) + + require.NoError(t, s.MainDB.Create(&database.Webhook{ + BaseModel: database.BaseModel{ID: s.WebhookID}, + UserID: uuid.New().String(), + Name: name, + }).Error) + + return s +} + +// fDrain collects every task the engine has queued. +// +// Every caller drives the dispatch paths synchronously and has already +// waited for them to return, so anything they queued is in the channel +// by now. The short grace covers nothing but scheduler jitter, and is +// kept small because one of these tests runs the drain forty times. +func fDrain(e *delivery.Engine) []delivery.Task { + var out []delivery.Task + + for { + select { + case task := <-e.ExportDeliveryCh(): + out = append(out, task) + case task := <-e.ExportRetryCh(): + out = append(out, task) + case <-time.After(25 * time.Millisecond): + return out + } + } +} + +// TestArchiveHandleIsWAL closes the last gap in the durability +// evidence: the main and per-webhook tiers each assert their journal +// mode on a live handle, and the archive tier gets its settings from +// the same code path but nothing checked the running file. +func TestArchiveHandleIsWAL(t *testing.T) { + t.Parallel() + + w := delivery.NewExportArchiveWriter( + filepath.Join(t.TempDir(), "archive-wal.db"), + archiveTestLogger(), 0, + ) + + require.NoError(t, w.Open(0)) + + var mode string + + row := w.DB().Raw("pragma journal_mode").Row() + require.NoError(t, row.Scan(&mode)) + assert.Equal(t, "wal", strings.ToLower(mode)) + + var busy string + + row = w.DB().Raw("pragma busy_timeout").Row() + require.NoError(t, row.Scan(&busy)) + assert.Equal(t, "10000", busy) +} + +// TestSweepLeavesAQueuedDeliveryAlone is the case the age bound cannot +// see. The delivery is queued and untouched, so its row is arbitrarily +// old and still perfectly healthy; only ownership distinguishes it +// from a stranded one. +func TestSweepLeavesAQueuedDeliveryAlone(t *testing.T) { + t.Parallel() + + targetID := uuid.New().String() + s := fSweepSetup(t, targetID, "queued") + + event := iSeedEvent( + t, s.WebhookDB, s.WebhookID, `{"queued":true}`, + ) + + d := iSeedDelivery( + t, s.WebhookDB, event.ID, targetID, + database.DeliveryStatusPending, + ) + rAgePending(t, s.WebhookDB, d.ID) + + // Queued exactly as the receiver queues it, and never dequeued: + // no workers are running in this engine. + s.Engine.Notify([]delivery.Task{{ + DeliveryID: d.ID, + EventID: event.ID, + WebhookID: s.WebhookID, + TargetID: targetID, + }}) + + require.Equal(t, 1, s.Engine.ExportInflightHeld()) + + s.Engine.ExportSweepWebhookRetries( + context.Background(), s.WebhookID, + ) + + tasks := fDrain(s.Engine) + assert.Len( + t, tasks, 1, + "the sweep must not queue a delivery that is "+ + "already waiting for a worker", + ) +} + +// TestRecoveryAndSweepDoNotDoubleDispatch drives the two entry points +// the engine starts concurrently against one aged pending row. Before +// ownership they both dispatched it. +func TestRecoveryAndSweepDoNotDoubleDispatch(t *testing.T) { + t.Parallel() + + targetID := uuid.New().String() + s := fSweepSetup(t, targetID, "racing") + + event := iSeedEvent( + t, s.WebhookDB, s.WebhookID, `{"racing":true}`, + ) + + ctx := context.Background() + + for range 40 { + d := iSeedDelivery( + t, s.WebhookDB, event.ID, targetID, + database.DeliveryStatusPending, + ) + rAgePending(t, s.WebhookDB, d.ID) + + var wg sync.WaitGroup + + wg.Go(func() { + s.Engine.ExportRecoverPendingDeliveries( + ctx, s.WebhookDB, s.WebhookID, + ) + }) + wg.Go(func() { + s.Engine.ExportSweepWebhookRetries( + ctx, s.WebhookID, + ) + }) + wg.Wait() + + tasks := fDrain(s.Engine) + require.Len( + t, tasks, 1, + "delivery %s dispatched %d times", + d.ID, len(tasks), + ) + + // No worker runs in this engine, so the reference the winner + // took is never released and earlier iterations' deliveries + // stay owned — which is itself the property under test, since + // both paths see them on every subsequent pass. + } +} + +// TestConcurrentClaimsOfOneDeliveryYieldOneOwner exercises the +// exclusion directly, rather than arguing it from a SQL predicate. +func TestConcurrentClaimsOfOneDeliveryYieldOneOwner( + t *testing.T, +) { + t.Parallel() + + eng := newISetup(t).Engine + deliveryID := uuid.New().String() + + var ( + wg sync.WaitGroup + mu sync.Mutex + won int + ) + + for range 64 { + wg.Go(func() { + if eng.ExportRetainDelivery(deliveryID) { + mu.Lock() + won++ + mu.Unlock() + } + }) + } + + wg.Wait() + + assert.Equal(t, 1, won) + assert.Equal(t, 1, eng.ExportInflightHeld()) +} + +// TestOwnershipIsReleasedAfterDelivery guards the other direction: a +// leaked reference hides a delivery from every sweep for the life of +// the process. +func TestOwnershipIsReleasedAfterDelivery(t *testing.T) { + t.Parallel() + + s := newISetup(t) + targetID := uuid.New().String() + + iCreateTarget(t, s.MainDB, targetID, + s.WebhookID, "released", + database.TargetTypeLog, "", 0, + ) + + event := iSeedEvent( + t, s.WebhookDB, s.WebhookID, `{"released":true}`, + ) + + d := iSeedDelivery( + t, s.WebhookDB, event.ID, targetID, + database.DeliveryStatusPending, + ) + + s.Engine.ExportStart() + + defer func() { + require.NoError( + t, s.Engine.ExportStop(context.Background()), + ) + }() + + body := `{"released":true}` + + s.Engine.Notify([]delivery.Task{{ + DeliveryID: d.ID, + EventID: event.ID, + WebhookID: s.WebhookID, + TargetID: targetID, + TargetName: "released", + TargetType: database.TargetTypeLog, + Body: &body, + EntrypointID: event.EntrypointID, + }}) + + iWaitForDelivered(t, s.WebhookDB, d.ID) + + assert.Eventually( + t, + func() bool { + return s.Engine.ExportInflightHeld() == 0 + }, + 2*time.Second, 20*time.Millisecond, + "the delivery stayed owned after it was delivered", + ) +} + +// TestRetryingRecoverySkipsASuccessfulResult is the retrying-side twin +// of the pending reconcile. A second attempt that reached the receiver +// and whose status write then failed sits at retrying holding a +// successful result, and re-sending it is the same duplicate. +func TestRetryingRecoverySkipsASuccessfulResult(t *testing.T) { + t.Parallel() + + targetID := uuid.New().String() + s := fSweepSetup(t, targetID, "retry-settled") + + event := iSeedEvent( + t, s.WebhookDB, s.WebhookID, `{"retry":true}`, + ) + + d := iSeedDelivery( + t, s.WebhookDB, event.ID, targetID, + database.DeliveryStatusRetrying, + ) + rSeedResult(t, s.WebhookDB, d.ID, 1, false) + rSeedResult(t, s.WebhookDB, d.ID, 2, true) + + s.Engine.ExportRecoverRetryingDeliveries( + s.WebhookDB, s.WebhookID, + ) + + assert.Empty( + t, fDrain(s.Engine), + "a retrying delivery holding a successful result "+ + "must not be sent again", + ) + + iAssertStatus( + t, s.WebhookDB, d.ID, + database.DeliveryStatusDelivered, + ) +} + +// TestRetryingSweepSkipsASuccessfulResult is the same rule on the +// periodic sweep's retrying arm. +func TestRetryingSweepSkipsASuccessfulResult(t *testing.T) { + t.Parallel() + + targetID := uuid.New().String() + s := fSweepSetup(t, targetID, "retry-swept") + + event := iSeedEvent( + t, s.WebhookDB, s.WebhookID, `{"swept":true}`, + ) + + d := iSeedDelivery( + t, s.WebhookDB, event.ID, targetID, + database.DeliveryStatusRetrying, + ) + rSeedResult(t, s.WebhookDB, d.ID, 1, false) + rSeedResult(t, s.WebhookDB, d.ID, 2, true) + + s.Engine.ExportSweepWebhookRetries( + context.Background(), s.WebhookID, + ) + + assert.Empty(t, fDrain(s.Engine)) + + iAssertStatus( + t, s.WebhookDB, d.ID, + database.DeliveryStatusDelivered, + ) + + var attempts int64 + + require.NoError(t, s.WebhookDB. + Model(&database.DeliveryResult{}). + Where("delivery_id = ?", d.ID). + Count(&attempts).Error) + assert.Equal( + t, int64(2), attempts, + "settling must not invent an attempt", + ) +} + +// TestScheduledRetryIsNotSweptDuringBackoff closes the window between +// a target scheduling a retry and the timer firing. The row says +// retrying and nothing is running, which is exactly what an orphaned +// retry looks like from the database. +func TestScheduledRetryIsNotSweptDuringBackoff(t *testing.T) { + t.Parallel() + + targetID := uuid.New().String() + s := fSweepSetup(t, targetID, "backoff") + + event := iSeedEvent( + t, s.WebhookDB, s.WebhookID, `{"backoff":true}`, + ) + + d := iSeedDelivery( + t, s.WebhookDB, event.ID, targetID, + database.DeliveryStatusRetrying, + ) + + s.Engine.ExportScheduleRetry(delivery.Task{ + DeliveryID: d.ID, + EventID: event.ID, + WebhookID: s.WebhookID, + TargetID: targetID, + AttemptNum: 2, + }, time.Hour) + + require.Equal(t, 1, s.Engine.ExportInflightHeld()) + + s.Engine.ExportSweepWebhookRetries( + context.Background(), s.WebhookID, + ) + + assert.Empty( + t, fDrain(s.Engine), + "the sweep must not duplicate a retry that is "+ + "already scheduled", + ) +} + +// TestRedispatchStampsTheRow pins the cadence control: a stranded +// delivery that has just been handed out is not selected again by the +// next tick a minute later. +func TestRedispatchStampsTheRow(t *testing.T) { + t.Parallel() + + targetID := uuid.New().String() + s := fSweepSetup(t, targetID, "stamped") + + event := iSeedEvent( + t, s.WebhookDB, s.WebhookID, `{"stamped":true}`, + ) + + d := iSeedDelivery( + t, s.WebhookDB, event.ID, targetID, + database.DeliveryStatusPending, + ) + rAgePending(t, s.WebhookDB, d.ID) + + ctx := context.Background() + + s.Engine.ExportSweepWebhookRetries(ctx, s.WebhookID) + require.Len(t, fDrain(s.Engine), 1) + + var row database.Delivery + + require.NoError(t, s.WebhookDB. + First(&row, "id = ?", d.ID).Error) + assert.WithinDuration( + t, time.Now(), row.UpdatedAt, time.Minute, + "a re-dispatched delivery must be stamped so the "+ + "next tick does not select it again", + ) +} diff --git a/internal/delivery/queue_depth_gormlog_test.go b/internal/delivery/queue_depth_gormlog_test.go index e733990..8c89734 100644 --- a/internal/delivery/queue_depth_gormlog_test.go +++ b/internal/delivery/queue_depth_gormlog_test.go @@ -3,8 +3,6 @@ package delivery_test import ( "bytes" "context" - "database/sql" - "fmt" "log/slog" "net/http" "path/filepath" @@ -54,12 +52,10 @@ func (q *qdSyncBuf) String() string { func qdMainDB(t *testing.T, log *slog.Logger) *gorm.DB { t.Helper() - dsn := fmt.Sprintf( - "file:%s?cache=shared&mode=rwc", + sqlDB, err := database.OpenSQLite( filepath.Join(t.TempDir(), "main-gormlog.db"), + database.SQLiteModeCreate, ) - - sqlDB, err := sql.Open("sqlite", dsn) require.NoError(t, err) t.Cleanup(func() { _ = sqlDB.Close() }) diff --git a/internal/delivery/recovery_durability_test.go b/internal/delivery/recovery_durability_test.go new file mode 100644 index 0000000..a139eeb --- /dev/null +++ b/internal/delivery/recovery_durability_test.go @@ -0,0 +1,378 @@ +package delivery_test + +import ( + "context" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "gorm.io/gorm" + "sneak.berlin/go/webhooker/internal/database" + "sneak.berlin/go/webhooker/internal/delivery" +) + +// These tests cover the delivery half of +// https://git.eeqj.de/sneak/webhooker/issues/256: a delivery that +// reached its receiver but whose bookkeeping write failed used to be +// left at pending and re-sent on the next restart, giving the receiver +// a second copy while the event log recorded one attempt. + +// rSeedResult records a DeliveryResult against a delivery, standing in +// for the attempt row the send path writes before the status. +func rSeedResult( + t *testing.T, + db *gorm.DB, + deliveryID string, + attemptNum int, + success bool, +) { + t.Helper() + + require.NoError(t, db.Create(&database.DeliveryResult{ + DeliveryID: deliveryID, + AttemptNum: attemptNum, + Success: success, + }).Error) +} + +// rAgePending backdates a delivery past the sweep's age bound, which is +// what separates a stranded delivery from one a worker still holds. +func rAgePending( + t *testing.T, db *gorm.DB, deliveryID string, +) { + t.Helper() + + old := time.Now().Add( + -2 * delivery.ExportPendingSweepMinAge, + ) + + require.NoError(t, db.Model(&database.Delivery{}). + Where("id = ?", deliveryID). + UpdateColumn("updated_at", old).Error) +} + +func TestRecoverySkipsPendingWithSuccessfulResult( + t *testing.T, +) { + t.Parallel() + + s := newISetup(t) + targetID := uuid.New().String() + + iCreateTarget(t, s.MainDB, targetID, + s.WebhookID, "already-delivered", + database.TargetTypeLog, "", 0, + ) + + event := iSeedEvent( + t, s.WebhookDB, s.WebhookID, `{"delivered":true}`, + ) + + // The delivery whose send succeeded and whose result row landed: + // only the status write failed, so it sits at pending. + done := iSeedDelivery( + t, s.WebhookDB, event.ID, targetID, + database.DeliveryStatusPending, + ) + rSeedResult(t, s.WebhookDB, done.ID, 1, true) + + // A delivery that was genuinely never attempted. + fresh := iSeedDelivery( + t, s.WebhookDB, event.ID, targetID, + database.DeliveryStatusPending, + ) + + s.Engine.ExportRecoverPendingDeliveries( + context.Background(), s.WebhookDB, s.WebhookID, + ) + + select { + case task := <-s.Engine.ExportDeliveryCh(): + assert.Equal( + t, fresh.ID, task.DeliveryID, + "only the unattempted delivery may be re-sent", + ) + case <-time.After(2 * time.Second): + t.Fatal("expected the unattempted delivery") + } + + select { + case task := <-s.Engine.ExportDeliveryCh(): + t.Fatalf( + "re-sent an already delivered delivery: %s", + task.DeliveryID, + ) + case <-time.After(200 * time.Millisecond): + } + + // It is settled rather than merely skipped: leaving it pending + // would strand it again on the next sweep. + iAssertStatus( + t, s.WebhookDB, done.ID, + database.DeliveryStatusDelivered, + ) +} + +// TestRecoveryContinuesTheAttemptNumbering pins the audit trail: a +// recovered delivery that already recorded two attempts is re-sent as +// attempt three, not as attempt one again. +func TestRecoveryContinuesTheAttemptNumbering(t *testing.T) { + t.Parallel() + + s := newISetup(t) + targetID := uuid.New().String() + + iCreateTarget(t, s.MainDB, targetID, + s.WebhookID, "numbering", + database.TargetTypeLog, "", 0, + ) + + event := iSeedEvent( + t, s.WebhookDB, s.WebhookID, `{"numbering":true}`, + ) + + d := iSeedDelivery( + t, s.WebhookDB, event.ID, targetID, + database.DeliveryStatusPending, + ) + + rSeedResult(t, s.WebhookDB, d.ID, 1, false) + rSeedResult(t, s.WebhookDB, d.ID, 2, false) + + s.Engine.ExportRecoverPendingDeliveries( + context.Background(), s.WebhookDB, s.WebhookID, + ) + + select { + case task := <-s.Engine.ExportDeliveryCh(): + assert.Equal(t, d.ID, task.DeliveryID) + assert.Equal(t, 3, task.AttemptNum) + case <-time.After(2 * time.Second): + t.Fatal("expected the delivery to be recovered") + } +} + +// TestSweepRecoversStrandedPending is the half that removes the +// restart requirement: a delivery left at pending is picked up by the +// periodic sweep. +func TestSweepRecoversStrandedPending(t *testing.T) { + t.Parallel() + + targetID := uuid.New().String() + s := fSweepSetup(t, targetID, "stranded") + + event := iSeedEvent( + t, s.WebhookDB, s.WebhookID, `{"stranded":true}`, + ) + + stranded := iSeedDelivery( + t, s.WebhookDB, event.ID, targetID, + database.DeliveryStatusPending, + ) + rAgePending(t, s.WebhookDB, stranded.ID) + + // A delivery a worker may still be holding: young, and therefore + // none of the sweep's business. + inFlight := iSeedDelivery( + t, s.WebhookDB, event.ID, targetID, + database.DeliveryStatusPending, + ) + + s.Engine.ExportSweepWebhookRetries( + context.Background(), s.WebhookID, + ) + + select { + case task := <-s.Engine.ExportDeliveryCh(): + assert.Equal(t, stranded.ID, task.DeliveryID) + case <-time.After(2 * time.Second): + t.Fatal("expected the stranded delivery") + } + + select { + case task := <-s.Engine.ExportDeliveryCh(): + t.Fatalf( + "swept an in-flight delivery: %s", + task.DeliveryID, + ) + case <-time.After(200 * time.Millisecond): + } + + iAssertStatus( + t, s.WebhookDB, inFlight.ID, + database.DeliveryStatusPending, + ) +} + +// TestSweepClaimsAStrandedDeliveryOnlyOnce guards the repeat the sweep +// would otherwise be: the row stays pending for as long as the attempt +// runs, and a sweep a minute later must not send it a second time. +func TestSweepClaimsAStrandedDeliveryOnlyOnce(t *testing.T) { + t.Parallel() + + targetID := uuid.New().String() + s := fSweepSetup(t, targetID, "claimed") + + event := iSeedEvent( + t, s.WebhookDB, s.WebhookID, `{"claimed":true}`, + ) + + d := iSeedDelivery( + t, s.WebhookDB, event.ID, targetID, + database.DeliveryStatusPending, + ) + rAgePending(t, s.WebhookDB, d.ID) + + ctx := context.Background() + + s.Engine.ExportSweepWebhookRetries(ctx, s.WebhookID) + + select { + case task := <-s.Engine.ExportDeliveryCh(): + assert.Equal(t, d.ID, task.DeliveryID) + case <-time.After(2 * time.Second): + t.Fatal("expected the stranded delivery") + } + + // The delivery is still pending — nothing has run it yet — but + // the claim must keep the next sweep off it. + iAssertStatus( + t, s.WebhookDB, d.ID, + database.DeliveryStatusPending, + ) + + s.Engine.ExportSweepWebhookRetries(ctx, s.WebhookID) + + select { + case task := <-s.Engine.ExportDeliveryCh(): + t.Fatalf( + "sent a claimed delivery again: %s", + task.DeliveryID, + ) + case <-time.After(200 * time.Millisecond): + } +} + +// TestSweepSettlesStrandedPendingWithoutResending is the sweep's own +// version of the reconcile: a stranded delivery holding a successful +// result is settled where it stands, and the receiver hears nothing. +func TestSweepSettlesStrandedPendingWithoutResending( + t *testing.T, +) { + t.Parallel() + + targetID := uuid.New().String() + s := fSweepSetup(t, targetID, "settled") + + event := iSeedEvent( + t, s.WebhookDB, s.WebhookID, `{"settled":true}`, + ) + + d := iSeedDelivery( + t, s.WebhookDB, event.ID, targetID, + database.DeliveryStatusPending, + ) + rSeedResult(t, s.WebhookDB, d.ID, 1, true) + rAgePending(t, s.WebhookDB, d.ID) + + s.Engine.ExportSweepWebhookRetries( + context.Background(), s.WebhookID, + ) + + select { + case task := <-s.Engine.ExportDeliveryCh(): + t.Fatalf( + "re-sent a delivery that already succeeded: %s", + task.DeliveryID, + ) + case <-time.After(200 * time.Millisecond): + } + + iAssertStatus( + t, s.WebhookDB, d.ID, + database.DeliveryStatusDelivered, + ) + + var attempts int64 + + require.NoError(t, s.WebhookDB. + Model(&database.DeliveryResult{}). + Where("delivery_id = ?", d.ID). + Count(&attempts).Error) + assert.Equal( + t, int64(1), attempts, + "settling must not invent an attempt", + ) +} + +// TestFailedResultWriteLeavesDeliveryRecoverable is the rule the +// targets now follow: a bookkeeping write that fails must not advance +// the status, because pending and retrying are the states the sweeps +// recover and delivered is a claim the database refused to record. +func TestFailedResultWriteLeavesDeliveryRecoverable( + t *testing.T, +) { + t.Parallel() + + s := newISetup(t) + targetID := uuid.New().String() + + var hits atomic.Int64 + + ts := httptest.NewServer(http.HandlerFunc( + func(w http.ResponseWriter, _ *http.Request) { + hits.Add(1) + w.WriteHeader(http.StatusOK) + }, + )) + defer ts.Close() + + event := iSeedEvent( + t, s.WebhookDB, s.WebhookID, `{"unwritable":true}`, + ) + + d := iSeedDelivery( + t, s.WebhookDB, event.ID, targetID, + database.DeliveryStatusPending, + ) + + // Drop the table the attempt row goes in, so the send succeeds + // and only the bookkeeping write fails. + require.NoError( + t, + s.WebhookDB.Exec("drop table delivery_results").Error, + ) + + full := &database.Delivery{ + EventID: event.ID, + TargetID: targetID, + Status: database.DeliveryStatusPending, + Event: event, + Target: database.Target{ + Name: "unwritable", + Type: database.TargetTypeHTTP, + Config: iHTTPConfig(ts.URL), + }, + } + full.ID = d.ID + + s.Engine.ExportDeliverHTTP( + context.Background(), s.WebhookDB, full, + &delivery.Task{DeliveryID: d.ID, AttemptNum: 1}, + ) + + assert.Equal( + t, int64(1), hits.Load(), + "the send itself must still happen", + ) + + iAssertStatus( + t, s.WebhookDB, d.ID, + database.DeliveryStatusPending, + ) +} diff --git a/internal/delivery/target_database.go b/internal/delivery/target_database.go index 04ccaf6..d319551 100644 --- a/internal/delivery/target_database.go +++ b/internal/delivery/target_database.go @@ -58,12 +58,17 @@ func (t *databaseTarget) Deliver( "error", err, ) - t.eng.recordResult( + recErr := t.eng.recordResult( webhookDB, d, 1, false, 0, "", err.Error(), elapsed.Milliseconds(), ) + if recErr != nil { + t.eng.bookkeepingFailed(d, recErr) - t.eng.updateDeliveryStatus( + return + } + + t.eng.settleStatus( webhookDB, d, d.Target.Type, database.DeliveryStatusFailed, ) @@ -71,12 +76,17 @@ func (t *databaseTarget) Deliver( return } - t.eng.recordResult( + recErr := t.eng.recordResult( webhookDB, d, 1, true, 0, "", "", elapsed.Milliseconds(), ) + if recErr != nil { + t.eng.bookkeepingFailed(d, recErr) - t.eng.updateDeliveryStatus( + return + } + + t.eng.settleStatus( webhookDB, d, d.Target.Type, database.DeliveryStatusDelivered, ) diff --git a/internal/delivery/target_database_archive.go b/internal/delivery/target_database_archive.go index d179e07..0937854 100644 --- a/internal/delivery/target_database_archive.go +++ b/internal/delivery/target_database_archive.go @@ -1,7 +1,6 @@ package delivery import ( - "database/sql" "encoding/json" "errors" "fmt" @@ -12,6 +11,7 @@ import ( "gorm.io/driver/sqlite" "gorm.io/gorm" + "sneak.berlin/go/webhooker/internal/database" "sneak.berlin/go/webhooker/internal/gormlog" ) @@ -30,13 +30,13 @@ const ( // 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" + archiveModeCreate = database.SQLiteModeCreate // 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" + archiveModeExisting = database.SQLiteModeExisting ) var ( @@ -273,9 +273,11 @@ func (w *archiveWriter) open(expiry time.Duration) error { 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) + // Opened through database.OpenSQLite so an archive file carries + // the same WAL journaling, busy timeout, immediate-transaction + // locking, and pool bounds as every other database file. See + // internal/database/sqlite_open.go. + sqlDB, err := database.OpenSQLite(w.path, mode) if err != nil { return fmt.Errorf( "opening archive database %s: %w", w.path, err, diff --git a/internal/delivery/target_http.go b/internal/delivery/target_http.go index 93cdf3f..127c7c4 100644 --- a/internal/delivery/target_http.go +++ b/internal/delivery/target_http.go @@ -77,14 +77,19 @@ func (c *httpCore) fireAndForget( ) { c.eng.observeAttempt(d.Target.Type, res.elapsed()) - c.eng.recordResult( + err := c.eng.recordResult( webhookDB, d, 1, res.success, res.statusCode, res.respBody, res.errMsg, res.duration, ) + if err != nil { + c.eng.bookkeepingFailed(d, err) + + return + } if res.success { - c.eng.updateDeliveryStatus( + c.eng.settleStatus( webhookDB, d, d.Target.Type, database.DeliveryStatusDelivered, ) @@ -92,7 +97,7 @@ func (c *httpCore) fireAndForget( return } - c.eng.updateDeliveryStatus( + c.eng.settleStatus( webhookDB, d, d.Target.Type, database.DeliveryStatusFailed, ) @@ -122,16 +127,25 @@ func (c *httpCore) withRetry( c.eng.observeAttempt(d.Target.Type, res.elapsed()) - c.eng.recordResult( + err := c.eng.recordResult( webhookDB, d, attemptNum, res.success, res.statusCode, res.respBody, res.errMsg, res.duration, ) + if err != nil { + // The breaker still learns the outcome: it describes the + // target's health, which is unaffected by this database's. + c.recordCircuitOutcome(cb, res.success) + + c.eng.bookkeepingFailed(d, err) + + return + } if res.success { cb.RecordSuccess() - c.eng.updateDeliveryStatus( + c.eng.settleStatus( webhookDB, d, d.Target.Type, database.DeliveryStatusDelivered, ) @@ -146,6 +160,20 @@ func (c *httpCore) withRetry( ) } +// recordCircuitOutcome feeds one attempt's outcome to the target's +// circuit breaker. +func (c *httpCore) recordCircuitOutcome( + cb *CircuitBreaker, success bool, +) { + if success { + cb.RecordSuccess() + + return + } + + cb.RecordFailure() +} + func (c *httpCore) circuitBreakerBlock( webhookDB *gorm.DB, d *database.Delivery, @@ -169,7 +197,7 @@ func (c *httpCore) circuitBreakerBlock( "cooldown_remaining", remaining, ) - c.eng.updateDeliveryStatus( + c.eng.settleStatus( webhookDB, d, d.Target.Type, database.DeliveryStatusRetrying, ) @@ -189,7 +217,7 @@ func (c *httpCore) handleRetry( attemptNum int, ) { if attemptNum >= maxRetries { - c.eng.updateDeliveryStatus( + c.eng.settleStatus( webhookDB, d, d.Target.Type, database.DeliveryStatusFailed, ) @@ -197,7 +225,7 @@ func (c *httpCore) handleRetry( return } - c.eng.updateDeliveryStatus( + c.eng.settleStatus( webhookDB, d, d.Target.Type, database.DeliveryStatusRetrying, ) @@ -332,12 +360,17 @@ func (t *httpTarget) Deliver( "error", err, ) - t.eng.recordResult( + recErr := t.eng.recordResult( webhookDB, d, task.AttemptNum, false, 0, "", err.Error(), 0, ) + if recErr != nil { + t.eng.bookkeepingFailed(d, recErr) - t.eng.updateDeliveryStatus( + return + } + + t.eng.settleStatus( webhookDB, d, d.Target.Type, database.DeliveryStatusFailed, ) diff --git a/internal/delivery/target_log.go b/internal/delivery/target_log.go index 26aa7e2..096a5c7 100644 --- a/internal/delivery/target_log.go +++ b/internal/delivery/target_log.go @@ -55,12 +55,17 @@ func (t *logTarget) Deliver( t.eng.observeAttempt(d.Target.Type, elapsed) - t.eng.recordResult( + err := t.eng.recordResult( webhookDB, d, 1, true, 0, "", "", elapsed.Milliseconds(), ) + if err != nil { + t.eng.bookkeepingFailed(d, err) - t.eng.updateDeliveryStatus( + return + } + + t.eng.settleStatus( webhookDB, d, d.Target.Type, database.DeliveryStatusDelivered, ) diff --git a/internal/delivery/target_slack.go b/internal/delivery/target_slack.go index 3f855d1..f10cc15 100644 --- a/internal/delivery/target_slack.go +++ b/internal/delivery/target_slack.go @@ -95,12 +95,17 @@ func (t *slackTarget) failConfig( d *database.Delivery, err error, ) { - t.eng.recordResult( + recErr := t.eng.recordResult( webhookDB, d, 1, false, 0, "", err.Error(), 0, ) + if recErr != nil { + t.eng.bookkeepingFailed(d, recErr) - t.eng.updateDeliveryStatus( + return + } + + t.eng.settleStatus( webhookDB, d, d.Target.Type, database.DeliveryStatusFailed, ) diff --git a/internal/handlers/event_body.go b/internal/handlers/event_body.go index 3d862ff..d46b9a5 100644 --- a/internal/handlers/event_body.go +++ b/internal/handlers/event_body.go @@ -92,11 +92,10 @@ func (h *Handlers) HandleEventBodyDownload() http.HandlerFunc { // once per range. // // One consequence is worth keeping in view: the read finishes -// before the client is written to, so no read lock is held for -// the length of a slow download. These per-webhook databases -// run in SQLite's default journal mode rather than WAL, so a -// lock held that long would block the receiver from recording -// new events. +// before the client is written to, so nothing is held open for +// the length of a slow download. Under WAL a read no longer +// blocks the receiver, but it does pin the WAL against +// checkpointing, and a download can last minutes. func (h *Handlers) serveEventBody( w http.ResponseWriter, r *http.Request, diff --git a/internal/handlers/event_resubmit.go b/internal/handlers/event_resubmit.go index 6df139a..21e54b8 100644 --- a/internal/handlers/event_resubmit.go +++ b/internal/handlers/event_resubmit.go @@ -139,10 +139,10 @@ func (h *Handlers) resubmitEvent( } // Read before the write transaction is opened. The body can be up - // to the 1 MB ingest cap, and holding a read of it inside the - // transaction would extend how long the per-webhook database is - // locked against the receiver, which runs these files in - // SQLite's default journal mode rather than WAL. + // to the 1 MB ingest cap, and every transaction on these files + // takes the write lock at BEGIN (_txlock=immediate, see + // internal/database/sqlite_open.go), so reading inside it would + // hold that lock against the receiver for the length of the read. src, found, err := loadResubmitSource( webhookDB, webhook.ID, eventID.String(), )