Close archive writers when the delivery engine stops (closes #280)

The engine cached archive writers and never closed them at shutdown, so after a clean stop an archive's rows could sit in its -wal while the .db held no table. The engine's stop hook now evicts every cached writer once its workers have returned, the same way deleting a webhook does, so a clean stop leaves each archive as one file and a late write is refused. If the workers do not return within the stop budget, the writers are left open as a kill would leave them. The README no longer says archives keep their sidecars across a clean stop. Model: opus-5-5
2026-09-29 05:48:33 +00:00
parent d4f4ddf51f
commit 4452ef71fb
5 changed files with 173 additions and 20 deletions
@@ -362,6 +362,15 @@ func (e *Engine) start() {
 // stop cancels the worker pool's context and waits for the pool
 // to drain, bounded by the stop hook's context: a wedged worker
 // must not hang the process past fx's stop timeout.
+//
+// Once the pool has drained it closes the archive writers, so a
+// clean stop leaves no archive -wal behind. Nothing else holds a
+// writer for long by then: the archive sweeper stops before the
+// engine, and deleting a webhook only closes one. If the pool did
+// not drain in time, the writers are left open, as a kill would
+// leave them: a worker still running may be mid-write, and
+// closing its writer would wait on that write and then fail the
+// next delivery the worker archives.
 func (e *Engine) stop(ctx context.Context) error {
 	e.log.Info("delivery engine stopping")

@@ -376,6 +385,8 @@ func (e *Engine) stop(ctx context.Context) error {
 		return err
 	}

+	e.dbTarget.evictAll()
+
 	e.log.Info("delivery engine stopped")

 	return nil
@@ -2,6 +2,8 @@ package delivery_test

 import (
 	"context"
+	"fmt"
+	"path/filepath"
 	"testing"
 	"time"

@@ -269,3 +271,88 @@ func TestEngine_StopHookHonoursStopTimeout(t *testing.T) {

 	requireStopHookExpires(t, lc.hooks[0], "delivery engine")
 }
+
+// deliverToArchive runs one delivery to a database target through
+// the running engine and returns the webhook's archive file path.
+// The archive writer holds the file open afterwards.
+func deliverToArchive(t *testing.T, s iSetup) string {
+	t.Helper()
+
+	deliveryID, task := seedLogTask(t, s)
+	task.TargetType = database.TargetTypeDatabase
+
+	s.Engine.Notify([]delivery.Task{task})
+
+	iWaitForDelivered(t, s.WebhookDB, deliveryID)
+
+	return filepath.Join(
+		filepath.Dir(s.DBMgr.DBPath(s.WebhookID)),
+		fmt.Sprintf("archive-%s.db", s.WebhookID),
+	)
+}
+
+// TestEngine_StopHookClosesArchives is the regression test for an
+// archive split across two files by a clean stop. The engine never
+// closed its archive writers, so after a stop the archived rows
+// could sit in archive-{id}.db-wal while archive-{id}.db held no
+// table at all, and copying the .db on its own gave an empty
+// database.
+func TestEngine_StopHookClosesArchives(t *testing.T) {
+	t.Parallel()
+
+	s := newISetup(t)
+
+	lc := startEngineViaHook(t, s.Engine)
+
+	path := deliverToArchive(t, s)
+	require.FileExists(
+		t, path+"-wal",
+		"an open archive should have a -wal for the stop to remove",
+	)
+
+	require.NoError(t, lc.hooks[0].OnStop(context.Background()))
+
+	wals, err := filepath.Glob(
+		filepath.Join(filepath.Dir(path), "archive-*.db-wal"),
+	)
+	require.NoError(t, err)
+	require.Empty(
+		t, wals, "a clean stop must leave no archive -wal behind",
+	)
+
+	// With no -wal beside it, the row can only be in the .db.
+	count, err := countArchivedRows(path)
+	require.NoError(t, err)
+	require.Equal(t, int64(1), count)
+}
+
+// TestEngine_StopHookTimeoutLeavesArchivesOpen covers a stop whose
+// budget runs out while a worker is still running. That worker may
+// be in the middle of an archive write, so the archive writers are
+// left open, as a kill would leave them, rather than closed
+// underneath it.
+func TestEngine_StopHookTimeoutLeavesArchivesOpen(t *testing.T) {
+	t.Parallel()
+
+	s := newISetup(t)
+
+	lc := startEngineViaHook(t, s.Engine)
+
+	deliverToArchive(t, s)
+
+	release := make(chan struct{})
+
+	t.Cleanup(func() {
+		close(release)
+		s.Engine.EvictWebhook(s.WebhookID)
+	})
+
+	s.Engine.ExportWedgeWorker(release)
+
+	requireStopHookExpires(t, lc.hooks[0], "delivery engine")
+
+	require.True(
+		t, s.Engine.ExportArchiveHandleOpen(s.WebhookID),
+		"a stop that timed out must not close archive writers",
+	)
+}
@@ -277,6 +277,24 @@ func (t *databaseTarget) evict(webhookID string) {
 	)
 }

+// evictAll evicts every cached archive writer, exactly as evict
+// does for one webhook. The engine calls it at shutdown, once its
+// workers have returned. Closing the last handle on an archive
+// moves the contents of its -wal into the .db and removes the
+// -wal, so a clean stop leaves each archive as a single file.
+func (t *databaseTarget) evictAll() {
+	t.mu.Lock()
+
+	writers := t.writers
+	t.writers = nil
+
+	t.mu.Unlock()
+
+	for _, w := range writers {
+		w.evict()
+	}
+}
+
 // 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
@@ -1,6 +1,7 @@
 package delivery_test

 import (
+	"context"
 	"errors"
 	"fmt"
 	"net/http"
@@ -361,3 +362,46 @@ func TestEvictWebhook_LaterDeliveryRecreatesWriter(t *testing.T) {
 		"a later delivery should recreate the writer",
 	)
 }
+
+// TestEngineStop_WriteAfterStopIsRefused proves the engine's stop
+// closes each archive writer the way deleting its webhook does: a
+// write that reaches a writer after the stop is refused, reopens
+// nothing and adds no row.
+func TestEngineStop_WriteAfterStopIsRefused(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)
+
+	w := eng.ExportArchiveWriterFor(event.WebhookID)
+	require.NotNil(t, w)
+	require.True(t, w.HandleOpen())
+
+	require.NoError(t, eng.ExportStop(context.Background()))
+
+	err := w.Write(evictTestRow("ev-after-stop"), 0)
+
+	require.ErrorIs(
+		t, err, delivery.ErrExportArchiveWriterEvicted,
+		"a write after the stop must be refused",
+	)
+	assert.False(
+		t, w.HandleOpen(),
+		"a refused write must not reopen the archive",
+	)
+	assert.False(
+		t, eng.ExportHasArchiveWriter(event.WebhookID),
+		"the stop should empty the registry",
+	)
+
+	count, err := countArchivedRows(w.Path())
+	require.NoError(t, err)
+	assert.Equal(
+		t, int64(1), count, "the refused row must not be written",
+	)
+}