Close archive writers when the delivery engine stops (closes #280)
check / check (push) Successful in 3m36s
check / check (push) Successful in 3m36s
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
This commit is contained in:
@@ -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",
|
||||
)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user