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