1 Commits

Author SHA1 Message Date
clawbot
190cabe0f2 Evict archive writers on deletion and sweep idle archives (closes #89)
All checks were successful
check / check (push) Successful in 3m10s
The per-webhook archiveWriter registry in the database delivery target
was never evicted, so a deleted webhook's writer -- and any archive
file handle open within its debounce window -- lingered for the process
lifetime. Separately, expiry pruning ran only when an archive was
(re)opened, and reopens only happen on writes, so an archive belonging
to a webhook that stopped receiving events kept its expired rows
forever.

Eviction: a new one-method delivery.WebhookEvictor interface (kept
separate from Notifier: archiving lifecycle is not notification) is
implemented by the Engine and injected into the handlers. Deleting a
webhook, or deleting its last database target, drops the writer from
the registry and closes its handle under the writer's own mutex, so
eviction can never race an in-flight write. An evicted writer refuses
further writes rather than reopening a file nothing holds. The archive
file is deliberately left on disk: it is long-term storage an operator
may want to keep or move away, and destroying it as a side effect of
deleting a webhook would be unrecoverable.

Idle sweep: a new ArchiveSweeper, modelled on the event RetentionReaper
(fx lifecycle hooks, cancellable context, WaitGroup, ticker loop),
prunes archives whose database target declares a positive expiry. It
reuses the existing RETENTION_SWEEP_INTERVAL rather than adding a
config key. It never creates an archive -- a missing file is skipped,
and the reopen uses SQLite mode=rw so the file cannot be conjured even
if it disappears mid-sweep -- routes the prune through the per-webhook
writer so its mutex orders the sweep against concurrent writes, and
leaves the archive closed so the move-the-file-away workflow keeps
working. A failure for one webhook is logged and the sweep continues.
Archives with no expiry or the expiry "never" are untouched.
2026-08-09 02:25:33 +00:00
10 changed files with 24 additions and 809 deletions

View File

@@ -544,12 +544,6 @@ afterwards so the move-the-file-away workflow keeps working. Archives
with no expiry, or the expiry `never`, are not touched by the sweep at
all.
Note that a webhook has one archive file but may carry more than one
`database` target, each with its own `expiry`. The shortest expiry
configured on any of them therefore governs the whole archive, and the
sweep applies it whether or not the webhook is still receiving events.
Configure a single `database` target per webhook unless you intend that.
Deleting a webhook releases its archive: the delivery engine's cached
archive writer is dropped and its file handle closed, so nothing lingers
after the webhook is gone. The archive **file itself is deliberately

View File

@@ -10,11 +10,9 @@
# Status
pre-1.0. No git tags exist. main (4f5ecb1) is a working webhook proxy
pre-1.0. No git tags exist. main (afe88c6) is a working webhook proxy
with auth, CSRF/SSRF protections, login rate limiting, Slack target,
event retention (#63), the database archiving target (#43), the admin
password change flow (#65), policy compliance (#6), and pinned lint
tooling (#55). Note: TODO.md was
policy compliance (#6), and pinned lint tooling (#55). Note: TODO.md was
deliberately deleted from this repo in f9a9569 (2026-03-01, #6); its
content was folded into the README TODO section, which this draft
reconstructs as of 2026-07-06.

View File

@@ -2,7 +2,6 @@ package delivery
import (
"context"
"errors"
"log/slog"
"sync"
"time"
@@ -61,22 +60,9 @@ func NewArchiveSweeper(
interval: params.Config.RetentionSweepInterval,
}
s.registerHooks(lc)
return s
}
// registerHooks wires the sweeper's start and stop into the fx
// lifecycle. Both hook contexts are deliberately ignored: see
// start for why the background loop must not inherit the start
// hook's context, and stop for why shutdown blocks on the loop
// rather than on the stop hook's deadline.
func (s *ArchiveSweeper) registerHooks(lc fx.Lifecycle) {
lc.Append(fx.Hook{
//nolint:contextcheck // Not passing the hook context is
// the point: see start.
OnStart: func(_ context.Context) error {
s.start()
OnStart: func(ctx context.Context) error {
s.start(ctx)
return nil
},
@@ -86,21 +72,12 @@ func (s *ArchiveSweeper) registerHooks(lc fx.Lifecycle) {
return nil
},
})
return s
}
// start launches the background sweep loop.
//
// The loop's context is derived from context.Background(), NOT
// from the fx OnStart hook context. The hook context carries
// fx's start timeout (15s by default), so a loop derived from it
// is cancelled 15 seconds after the application starts — long
// before the first tick under the default one-hour sweep
// interval, leaving a sweeper that never sweeps. A long-lived
// goroutine must outlive the startup phase, so its lifetime is
// bounded by OnStop instead: stop cancels this context and waits
// on the WaitGroup.
func (s *ArchiveSweeper) start() {
ctx, cancel := context.WithCancel(context.Background())
func (s *ArchiveSweeper) start(ctx context.Context) {
ctx, cancel := context.WithCancel(ctx)
s.cancel = cancel
s.wg.Add(1)
@@ -201,29 +178,12 @@ func (s *ArchiveSweeper) sweepTarget(target *database.Target) {
}
err = s.eng.dbTarget.sweepWebhook(target.WebhookID, expiry)
if err == nil {
return
}
// A writer evicted underneath the sweep means the operator
// deleted the webhook (or its last database target) while the
// sweep was walking the target list. That is an ordinary
// interleaving, not a failure, so it must not produce an
// error line.
if errors.Is(err, errArchiveWriterEvicted) {
s.log.Debug(
"archive sweep: writer evicted mid-sweep",
if err != nil {
s.log.Error(
"archive sweep: failed to prune archive",
"webhook_id", target.WebhookID,
"target_id", target.ID,
"error", err,
)
return
}
s.log.Error(
"archive sweep: failed to prune archive",
"webhook_id", target.WebhookID,
"target_id", target.ID,
"error", err,
)
}

View File

@@ -5,7 +5,6 @@ import (
"database/sql"
"fmt"
"net/http"
"os"
"path/filepath"
"sync"
"testing"
@@ -14,7 +13,6 @@ import (
"github.com/google/uuid"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"go.uber.org/fx"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
"gorm.io/gorm/clause"
@@ -194,292 +192,6 @@ func archivedEventIDs(
return ids
}
// countArchivedRows counts the rows in an archive file without
// asserting anything, so it is safe to poll from an
// assert.Eventually condition (which runs off the test
// goroutine, where testify assertions must not be used).
func countArchivedRows(path string) (int64, error) {
sqlDB, err := sql.Open(
"sqlite", fmt.Sprintf("file:%s?mode=ro", path),
)
if err != nil {
return 0, err
}
defer func() { _ = sqlDB.Close() }()
gdb, err := gorm.Open(
sqlite.Dialector{Conn: sqlDB}, &gorm.Config{},
)
if err != nil {
return 0, err
}
var count int64
err = gdb.Model(&delivery.ExportArchivedEvent{}).
Count(&count).Error
if err != nil {
return 0, err
}
return count, nil
}
// captureLifecycle is a minimal fx.Lifecycle that records the
// hooks a component registers, so a test can invoke the real
// OnStart/OnStop functions with a context of its choosing.
type captureLifecycle struct {
hooks []fx.Hook
}
func (l *captureLifecycle) Append(h fx.Hook) {
l.hooks = append(l.hooks, h)
}
// TestArchiveSweeper_LoopOutlivesStartHookContext is the
// regression test for a sweeper that never swept. fx calls
// OnStart with a context carrying the application's start
// timeout (15 seconds by default), so a background loop whose
// context is derived from it is cancelled 15 seconds into the
// process — three quarters of an hour before the first tick
// under the default one-hour sweep interval.
//
// The hook context here is already cancelled, which is the same
// defect taken to its limit: a loop that inherits it never runs
// a single tick, while a correctly rooted loop keeps sweeping
// for as long as the process lives. Handing the hook a plain
// context.Background() would assert nothing at all.
func TestArchiveSweeper_LoopOutlivesStartHookContext(
t *testing.T,
) {
t.Parallel()
env := setupSweeperTest(t)
webhookID := env.seedDatabaseTarget(t, `{"expiry":"1h"}`)
now := time.Now()
path := env.seedArchiveRows(
t, webhookID,
now.Add(-48*time.Hour),
now.Add(-time.Minute),
)
env.sweeper.ExportSetInterval(10 * time.Millisecond)
// Drive the genuine fx hooks the application registers,
// rather than a test-only entry point.
lc := &captureLifecycle{}
env.sweeper.ExportRegisterHooks(lc)
require.Len(t, lc.hooks, 1)
hookCtx, cancel := context.WithCancel(context.Background())
cancel()
require.NoError(t, lc.hooks[0].OnStart(hookCtx))
t.Cleanup(func() {
_ = lc.hooks[0].OnStop(context.Background())
})
assert.Eventually(
t,
func() bool {
count, err := countArchivedRows(path)
return err == nil && count == 1
},
5*time.Second,
10*time.Millisecond,
"the sweep loop must keep running after the start "+
"hook's context is done; it pruned nothing, so it "+
"inherited the hook context and died",
)
}
// TestArchiveSweep_DoesNotResurrectEvictedWriter covers the
// interleaving where a sweep tick has already listed a webhook's
// target when the webhook is deleted and its writer evicted. The
// sweep must not put a writer back into the registry: nothing
// would ever evict it again, which is precisely the leak this
// change exists to close.
func TestArchiveSweep_DoesNotResurrectEvictedWriter(
t *testing.T,
) {
t.Parallel()
env := setupSweeperTest(t)
webhookID := env.seedDatabaseTarget(t, `{"expiry":"1h"}`)
env.seedArchiveRows(
t, webhookID, time.Now().Add(-48*time.Hour),
)
// Prime the registry the way a delivery would, then evict as
// the deletion path does. The target row is deliberately left
// in place: this is the tick that listed the webhook before
// the deletion committed.
_, err := env.eng.ExportEnsureArchiveWriter(webhookID)
require.NoError(t, err)
env.eng.EvictWebhook(webhookID)
require.False(t, env.eng.ExportHasArchiveWriter(webhookID))
env.sweeper.ExportSweep(context.Background())
assert.False(
t, env.eng.ExportHasArchiveWriter(webhookID),
"a sweep must never re-register a writer for a webhook "+
"whose registry entry has already been released",
)
}
// TestArchiveSweep_LeavesNoRegistryEntry states the same
// invariant in its general form: sweeping an archive whose
// webhook has no cached writer must not leave one behind, so the
// registry keeps holding only writers a delivery created and an
// eviction can reach.
func TestArchiveSweep_LeavesNoRegistryEntry(t *testing.T) {
t.Parallel()
env := setupSweeperTest(t)
webhookID := env.seedDatabaseTarget(t, `{"expiry":"1h"}`)
path := env.seedArchiveRows(
t, webhookID,
time.Now().Add(-48*time.Hour),
time.Now().Add(-time.Minute),
)
require.False(t, env.eng.ExportHasArchiveWriter(webhookID))
env.sweeper.ExportSweep(context.Background())
assert.Equal(
t, []string{sweepRowNew}, archivedEventIDs(t, path),
"the sweep must still prune an idle archive",
)
assert.False(
t, env.eng.ExportHasArchiveWriter(webhookID),
"the sweep must release the registry entry it created",
)
}
// TestArchiveSweep_KeepsWriterAdoptedByDelivery is the other
// half of that invariant: an entry the sweep created but a
// delivery then claimed belongs to the registry and must survive
// the sweep, or the delivery would be left holding a detached
// writer with an open handle that no eviction can reach.
func TestArchiveSweep_KeepsWriterAdoptedByDelivery(
t *testing.T,
) {
t.Parallel()
env := setupSweeperTest(t)
webhookID := env.seedDatabaseTarget(t, `{"expiry":"1h"}`)
env.seedArchiveRows(
t, webhookID, time.Now().Add(-48*time.Hour),
)
webhookDB := testWebhookDB(t)
event := seedEvent(t, webhookDB, `{"n":1}`)
event.WebhookID = webhookID
d := seedDatabaseTargetDelivery(
t, webhookDB, event, `{"expiry":"1h"}`,
)
env.sweeper.ExportSweep(context.Background())
require.False(t, env.eng.ExportHasArchiveWriter(webhookID))
env.eng.ExportDeliverDatabase(webhookDB, d)
assert.True(
t, env.eng.ExportHasArchiveWriter(webhookID),
"a delivery's writer must stay registered",
)
env.sweeper.ExportSweep(context.Background())
assert.True(
t, env.eng.ExportHasArchiveWriter(webhookID),
"a sweep must not drop a writer a delivery owns",
)
}
// TestArchiveSweep_ContinuesAfterPerWebhookFailure proves a
// failure for one webhook does not abort the sweep for the
// others: an unparseable expiry and an unreadable archive both
// have to be logged and stepped over.
func TestArchiveSweep_ContinuesAfterPerWebhookFailure(
t *testing.T,
) {
t.Parallel()
env := setupSweeperTest(t)
// Seeded first so the sweep reaches them before the healthy
// webhook: targets come back in insertion order.
badConfigID := env.seedDatabaseTarget(t, `{"expiry":"!!!"}`)
env.seedArchiveRows(
t, badConfigID, time.Now().Add(-48*time.Hour),
)
corruptID := env.seedDatabaseTarget(t, `{"expiry":"1h"}`)
require.NoError(t, os.WriteFile(
env.archivePath(corruptID),
[]byte("this is not a sqlite database"),
0o600,
))
healthyID := env.seedDatabaseTarget(t, `{"expiry":"1h"}`)
healthyPath := env.seedArchiveRows(
t, healthyID,
time.Now().Add(-48*time.Hour),
time.Now().Add(-time.Minute),
)
env.sweeper.ExportSweep(context.Background())
assert.Equal(
t, []string{sweepRowNew},
archivedEventIDs(t, healthyPath),
"a failure for an earlier webhook must not stop the "+
"sweep from pruning the ones after it",
)
}
// TestArchiveSweep_OpenExistingDoesNotCreateFile pins the second
// of the two no-create guards. The first is the stat in
// sweepWebhook; this one is the SQLite open mode, which is what
// protects the window between that stat and the open. Flipping
// the sweep's mode to create-if-missing makes this fail.
func TestArchiveSweep_OpenExistingDoesNotCreateFile(
t *testing.T,
) {
t.Parallel()
dir := t.TempDir()
path := filepath.Join(dir, "archive-absent.db")
w := delivery.NewExportArchiveWriter(
path, archiveTestLogger(), 0,
)
err := w.OpenExisting(time.Hour)
require.Error(
t, err,
"opening a missing archive without create permission "+
"must fail rather than conjure the file",
)
for _, suffix := range archiveFileSuffixes() {
assert.NoFileExists(t, path+suffix)
}
}
// TestArchiveSweep_PrunesIdleArchive is the core regression
// test for this issue: an archive that receives no further
// writes must still lose its expired rows. Before the sweeper
@@ -578,7 +290,7 @@ func TestArchiveSweep_DoesNotCreateArchiveFile(t *testing.T) {
env.sweeper.ExportSweep(context.Background())
for _, suffix := range archiveFileSuffixes() {
for _, suffix := range []string{"", "-wal", "-shm"} {
assert.NoFileExists(
t, path+suffix,
"the sweep must not create an archive file",
@@ -705,7 +417,7 @@ func TestArchiveSweeper_StopsCleanly(t *testing.T) {
)
env.sweeper.ExportSetInterval(time.Millisecond)
env.sweeper.ExportStart()
env.sweeper.ExportStart(context.Background())
// stop blocks on the loop's WaitGroup, so returning at all
// proves the loop observed the cancellation and exited.

View File

@@ -7,17 +7,10 @@ import (
"net/http"
"time"
"go.uber.org/fx"
"gorm.io/gorm"
"sneak.berlin/go/webhooker/internal/database"
)
// ErrExportArchiveWriterEvicted exposes the sentinel returned by
// an evicted archive writer. It carries the Err prefix rather
// than this file's usual Export one because it is a sentinel
// error.
var ErrExportArchiveWriterEvicted = errArchiveWriterEvicted
// Exported constants for test access.
const (
ExportDeliveryChannelSize = deliveryChannelSize
@@ -335,59 +328,6 @@ func (e *ExportArchiveWriter) DB() *gorm.DB {
return e.w.db
}
// Path returns the archive file the writer owns.
func (e *ExportArchiveWriter) Path() string {
return e.w.path
}
// OpenExisting opens the archive without permitting creation,
// the way the idle sweep does.
func (e *ExportArchiveWriter) OpenExisting(
expiry time.Duration,
) error {
return e.w.openMode(archiveModeExisting, expiry)
}
// SweepExpired runs an idle sweep of the archive.
func (e *ExportArchiveWriter) SweepExpired(
expiry time.Duration,
) error {
return e.w.sweepExpired(expiry)
}
// Evict marks the writer evicted and closes its handle, exactly
// as leaving the registry does.
func (e *ExportArchiveWriter) Evict() {
e.w.evict()
}
// HandleOpen reports whether the writer currently holds an open
// archive handle.
func (e *ExportArchiveWriter) HandleOpen() bool {
e.w.mu.Lock()
defer e.w.mu.Unlock()
return e.w.db != nil
}
// ExportArchiveWriterFor returns the archive writer the registry
// currently caches for a webhook, or nil when none is cached. It
// never creates one, so a test can hold a reference to the very
// writer an eviction is about to detach.
func (e *Engine) ExportArchiveWriterFor(
webhookID string,
) *ExportArchiveWriter {
e.dbTarget.mu.Lock()
defer e.dbTarget.mu.Unlock()
w, ok := e.dbTarget.writers[webhookID]
if !ok {
return nil
}
return &ExportArchiveWriter{w: w}
}
// ExportHasArchiveWriter reports whether the database target
// currently caches an archive writer for a webhook.
func (e *Engine) ExportHasArchiveWriter(
@@ -458,16 +398,8 @@ func (s *ArchiveSweeper) ExportSweep(ctx context.Context) {
}
// ExportStart starts the sweeper's background loop for tests.
func (s *ArchiveSweeper) ExportStart() {
s.start()
}
// ExportRegisterHooks registers the sweeper's real fx lifecycle
// hooks on a lifecycle supplied by a test, so a test can drive
// the exact OnStart/OnStop functions the application runs and
// hand OnStart the kind of context fx actually supplies.
func (s *ArchiveSweeper) ExportRegisterHooks(lc fx.Lifecycle) {
s.registerHooks(lc)
func (s *ArchiveSweeper) ExportStart(ctx context.Context) {
s.start(ctx)
}
// ExportStop stops the sweeper's background loop for tests.

View File

@@ -130,78 +130,9 @@ func (t *databaseTarget) writerFor(
t.writers[webhookID] = w
}
// A delivery claims the entry: even if the idle sweep created
// it moments ago, it now belongs to the registry proper and
// the sweep must leave it in place when it finishes.
w.sweepOwned = false
return w, nil
}
// sweepWriterFor returns the archive writer the idle sweep should
// prune a webhook through, together with whether the sweep itself
// created the registry entry.
//
// The sweep must route its prune through the registered writer so
// the writer's mutex orders it against concurrent writes, but it
// must never leave a registry entry behind: a sweep that ran
// concurrently with the webhook's deletion would otherwise
// re-create an entry that nothing will ever evict again, which is
// exactly the leak eviction exists to prevent. An entry the sweep
// creates is therefore marked sweep-owned and handed back to
// releaseSweepWriter when the sweep is done.
func (t *databaseTarget) sweepWriterFor(
webhookID string,
) (*archiveWriter, bool, error) {
path, err := t.archivePath(webhookID)
if err != nil {
return nil, false, err
}
t.mu.Lock()
defer t.mu.Unlock()
if t.writers == nil {
t.writers = make(map[string]*archiveWriter)
}
w, ok := t.writers[webhookID]
if ok {
return w, false, nil
}
w = newArchiveWriter(path, t.eng.log)
w.sweepOwned = true
t.writers[webhookID] = w
return w, true, nil
}
// releaseSweepWriter drops a registry entry that the idle sweep
// created, so a sweep leaves the registry exactly as it found it.
//
// The entry is removed only if it is still the very writer the
// sweep installed and no delivery has claimed it in the meantime
// (writerFor clears sweepOwned when it hands a writer to the
// write path). Both conditions are evaluated under the registry
// lock, so an eviction that raced the sweep — which removes the
// entry outright — simply finds nothing left to do here, and a
// delivery that adopted the writer keeps a registered, evictable
// one.
func (t *databaseTarget) releaseSweepWriter(
webhookID string, w *archiveWriter,
) {
t.mu.Lock()
defer t.mu.Unlock()
cur, ok := t.writers[webhookID]
if !ok || cur != w || !cur.sweepOwned {
return
}
delete(t.writers, webhookID)
}
// archivePath returns the archive file path for a webhook: it
// lives beside the per-webhook event database in the data
// directory. It does not touch the filesystem.
@@ -262,11 +193,6 @@ func (t *databaseTarget) evict(webhookID string) {
// do) when the archive file does not exist, so a sweep never
// creates an archive for a webhook that has a database target
// but has never received an event.
//
// It also never leaves a registry entry behind: an entry it had
// to create to reach the writer's mutex is released again once
// the prune is done, so a sweep racing a webhook deletion cannot
// resurrect the writer the eviction just dropped.
func (t *databaseTarget) sweepWebhook(
webhookID string, expiry time.Duration,
) error {
@@ -282,14 +208,10 @@ func (t *databaseTarget) sweepWebhook(
return nil
}
w, created, err := t.sweepWriterFor(webhookID)
w, err := t.writerFor(webhookID)
if err != nil {
return err
}
if created {
defer t.releaseSweepWriter(webhookID, w)
}
return w.sweepExpired(expiry)
}

View File

@@ -190,19 +190,6 @@ type archiveWriter struct {
// never open the file again: nothing holds it any more, so a
// reopen would leak the handle for the process lifetime.
evicted bool
// sweepOwned marks a registry entry that the idle sweep
// created because no writer was cached for the webhook. The
// sweep removes such an entry again when it is done, so a
// sweep can never leave — or resurrect — a registry entry
// for a webhook that has been deleted. A delivery that adopts
// the writer clears the flag, handing the entry to the
// registry proper.
//
// Unlike every other field here it is guarded by
// databaseTarget.mu, not by this writer's mu: it describes the
// registry entry rather than the file.
sweepOwned bool
}
// newArchiveWriter builds an archiveWriter for a file path with

View File

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

View File

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

View File

@@ -257,59 +257,9 @@ func TestHandleTargetDelete_EvictsWhenLastDatabaseTargetGone(
)
}
// TestHandleTargetDelete_KeepsWriterWhenDatabaseTargetRemains
// proves that deleting one of several database targets leaves
// the still-needed archive writer alone: the surviving target
// keeps archiving to the same file, so the writer must stay.
func TestHandleTargetDelete_KeepsWriterWhenDatabaseTargetRemains(
t *testing.T,
) {
t.Parallel()
var (
h *handlers.Handlers
sess *session.Session
db *database.Database
ev *recordingEvictor
)
app := newTestApp(t, &h, &sess, &db, &ev)
app.RequireStart()
t.Cleanup(app.RequireStop)
wh := seedWebhook(t, db)
doomed := seedTarget(
t, db, wh.ID, database.TargetTypeDatabase,
)
seedTarget(t, db, wh.ID, database.TargetTypeDatabase)
cookies := authenticatedCookies(
t, sess, deleteTestUserID, deleteTestUsername,
)
req := postRequest(
"/source/"+wh.ID+"/targets/"+doomed.ID+"/delete",
cookies,
map[string]string{
paramSourceID: wh.ID,
paramTargetID: doomed.ID,
},
)
w := httptest.NewRecorder()
h.HandleTargetDelete().ServeHTTP(w, req)
require.Equal(t, http.StatusSeeOther, w.Code)
assert.Empty(
t, ev.Evicted(),
"a second database target still needs the writer",
)
}
// TestHandleTargetDelete_KeepsWriterWhileDatabaseTargetRemains
// proves that deleting an unrelated target type leaves a
// still-needed archive writer alone.
// proves that deleting an unrelated target, or one of several
// database targets, leaves a still-needed archive writer alone.
func TestHandleTargetDelete_KeepsWriterWhileDatabaseTargetRemains(
t *testing.T,
) {