Move the archive files into their own package (closes #541)
check / check (push) Waiting to run

The SQLite archive files the database target writes are now managed by
internal/archive: file names, rotation, the writer of each target,
pruning, renames, the gzipped JSON export, the idle sweep, and the
rename lock, which was Handlers.renameMu. The database target stays in
internal/delivery and writes through archive.Writers; the engine still
closes the writers when it stops, so the stop order is unchanged. The
handlers take the rename lock from archive.Archives.RenameLock, and
rename and evict through archive.Archives.

Exported names drop their Archive prefix in the new package. The tests
moved with the code; the two that check a delivery's recorded attempt
stay in delivery.

Model: opus-5-5
This commit was merged in pull request #552.
This commit is contained in:
2026-10-15 21:03:43 +02:00
parent f10440b7cc
commit 0cdda43e6c
42 changed files with 2178 additions and 1772 deletions
-253
View File
@@ -1,253 +0,0 @@
package delivery
import (
"context"
"errors"
"log/slog"
"sync"
"time"
"go.uber.org/fx"
"gorm.io/gorm"
"sneak.berlin/go/webhooker/internal/config"
"sneak.berlin/go/webhooker/internal/database"
"sneak.berlin/go/webhooker/internal/lifecycle"
"sneak.berlin/go/webhooker/internal/logger"
)
// ArchiveSweeperParams holds the fx dependencies for the
// ArchiveSweeper.
type ArchiveSweeperParams struct {
fx.In
Config *config.Config
Database *database.Database
Engine *Engine
Logger *logger.Logger
}
// ArchiveSweeper periodically prunes expired rows from the
// archive databases of database targets that carry a positive
// expiry.
//
// Without it, pruning happens only when an archive is
// (re)opened, and archives are only ever reopened by writes: an
// archive whose target has stopped receiving events would keep
// its expired rows forever. The sweep closes
// that gap without changing anything for archives whose expiry
// is unset or "never".
//
// It reuses Config.RetentionSweepInterval rather than
// introducing a second interval: this is a retention sweep with
// the same semantics as the event retention reaper.
type ArchiveSweeper struct {
db *database.Database
eng *Engine
log *slog.Logger
interval time.Duration
// cancel needs no lock: fx calls the stop hook only after the
// start hook has returned, so stop never reads it while start
// is still setting it.
cancel context.CancelFunc
wg sync.WaitGroup
}
// NewArchiveSweeper creates the archive sweeper and registers
// its fx lifecycle hooks. The background sweep loop starts on
// OnStart and stops cleanly on OnStop via context cancellation.
func NewArchiveSweeper(
lc fx.Lifecycle,
params ArchiveSweeperParams,
) *ArchiveSweeper {
s := &ArchiveSweeper{
db: params.Database,
eng: params.Engine,
log: params.Logger.Get(),
interval: params.Config.RetentionSweepInterval,
}
s.registerHooks(lc)
return s
}
// registerHooks wires the sweeper's start and stop into the fx
// lifecycle. The start hook's context is deliberately ignored
// (see start for why the background loop must not inherit it);
// the stop hook's context is honoured (see stop).
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()
return nil
},
OnStop: func(ctx context.Context) error {
return s.stop(ctx)
},
})
}
// 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())
s.cancel = cancel
s.wg.Add(1)
go s.run(ctx)
s.log.Info(
"archive sweeper started",
"interval", s.interval.String(),
)
}
// stop cancels the sweep loop's context and waits for it to
// exit, bounded by the stop hook's context: a prune wedged on a
// locked archive must not hang the process past fx's stop
// timeout.
func (s *ArchiveSweeper) stop(ctx context.Context) error {
s.log.Info("archive sweeper stopping")
if s.cancel != nil {
s.cancel()
}
err := lifecycle.WaitForShutdown(
ctx, s.log, "archive sweeper", &s.wg,
)
if err != nil {
return err
}
s.log.Info("archive sweeper stopped")
return nil
}
func (s *ArchiveSweeper) run(ctx context.Context) {
defer s.wg.Done()
ticker := time.NewTicker(s.interval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
s.sweep(ctx)
}
}
}
// sweep prunes every archive whose database target declares a
// positive expiry. Targets belonging to a deleted webhook are
// soft-deleted along with it, so GORM's default scope already
// excludes them.
//
// A failure for one target is logged and the sweep continues,
// matching how the write path already treats a prune error as
// non-fatal.
func (s *ArchiveSweeper) sweep(ctx context.Context) {
var targets []database.Target
err := s.db.DB().
WithContext(ctx).
Model(&database.Target{}).
Where("type = ?", database.TargetTypeDatabase).
Find(&targets).Error
if err != nil {
// The app stopping as a sweep starts cancels the listing.
// Stopping is not a failure, so it must not produce an
// error line.
if ctx.Err() != nil {
return
}
s.log.Error(
"archive sweep: failed to list database targets",
"error", err,
)
return
}
for i := range targets {
select {
case <-ctx.Done():
return
default:
}
s.sweepTarget(&targets[i])
}
}
// sweepTarget prunes the archive of a single database target.
// A missing, empty, or "never" expiry parses as a zero duration
// and is skipped entirely, so those archives are never pruned.
func (s *ArchiveSweeper) sweepTarget(target *database.Target) {
expiry, err := parseArchiveExpiry(target.Config)
if err != nil {
s.log.Error(
"archive sweep: invalid database target config",
"webhook_id", target.WebhookID,
"target_id", target.ID,
"error", err,
)
return
}
if expiry <= 0 {
return
}
if s.eng == nil || s.eng.dbTarget == nil {
return
}
err = s.eng.dbTarget.sweepArchive(target.ID, expiry)
if err == nil {
return
}
// A writer evicted, or a target row gone, underneath the sweep
// means the operator deleted the target or its webhook 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) ||
errors.Is(err, gorm.ErrRecordNotFound) {
s.log.Debug(
"archive sweep: target deleted mid-sweep",
"webhook_id", target.WebhookID,
"target_id", target.ID,
)
return
}
s.log.Error(
"archive sweep: failed to prune archive",
"webhook_id", target.WebhookID,
"target_id", target.ID,
"error", err,
)
}
File diff suppressed because it is too large Load Diff
+11 -68
View File
@@ -13,6 +13,7 @@ import (
"go.uber.org/fx"
"gorm.io/gorm"
"sneak.berlin/go/webhooker/internal/archive"
"sneak.berlin/go/webhooker/internal/database"
"sneak.berlin/go/webhooker/internal/globals"
"sneak.berlin/go/webhooker/internal/lifecycle"
@@ -124,31 +125,11 @@ type Notifier interface {
Notify(tasks []Task)
}
// Archives is how the handlers keep the database targets' archive
// files in step with the configuration. Deleting a webhook or a
// target releases the cached archive writers, whose open file
// handles would otherwise outlive them; renaming one renames the
// archive files, which are named for the webhook and the target
// (see ArchiveFileName).
//
// It is deliberately separate from Notifier: archiving lifecycle
// is not notification, and a small interface keeps the handlers
// package free of any dependency on the engine's internals while
// staying trivially fakeable in tests.
//
// Neither eviction deletes an archive file. Both are idempotent
// and are no-ops for a webhook or target with no engine state.
type Archives interface {
EvictWebhook(webhookID string)
EvictTarget(targetID string)
Rename(targetID, webhookName, targetName string) error
}
// CircuitBreakers is how the handlers read a target's circuit
// breaker, so the webhook page and the event log can say that
// deliveries to the target are paused and until when. Like Archives,
// it keeps the handlers free of the engine's internals and is
// trivially faked in tests.
// deliveries to the target are paused and until when. It keeps the
// handlers free of the engine's internals and is trivially faked in
// tests.
type CircuitBreakers interface {
StateAndCooldown(targetID string) (CircuitState, time.Duration)
}
@@ -164,6 +145,7 @@ type EngineParams struct {
Logger *logger.Logger
SSRFGuard *Guard
Metrics *metrics.Set
Archives *archive.Writers
}
// Engine processes queued deliveries in the background
@@ -202,9 +184,10 @@ type Engine struct {
httpTarget *httpTarget
slackTarget *slackTarget
// dbTarget is retained so the engine can reach the archive
// writer registry for eviction, renames and the idle sweep.
dbTarget *databaseTarget
// archives is the database targets' archive writers. The
// database target writes each event through them, and stop
// closes them.
archives *archive.Writers
// inflight is the set of deliveries this engine currently owns.
// Recovery and the sweeps re-dispatch only what it does not
@@ -228,6 +211,7 @@ func New(
workers: defaultWorkers,
version: params.Globals.Version,
mtr: params.Metrics,
archives: params.Archives,
}
e.initTargets(&http.Client{
@@ -272,47 +256,6 @@ func (e *Engine) Notify(tasks []Task) {
}
}
// EvictWebhook implements Archives. The cached archive writer of
// every database target of the webhook is dropped from the
// registry and its file handle closed. The archive files
// themselves are left on disk — they are long-term storage the
// operator owns.
func (e *Engine) EvictWebhook(webhookID string) {
if e.dbTarget == nil {
return
}
e.dbTarget.evictWebhook(webhookID)
}
// EvictTarget implements Archives. It is EvictWebhook for a single
// database target, and leaves the archive file on disk the same
// way.
func (e *Engine) EvictTarget(targetID string) {
if e.dbTarget == nil {
return
}
e.dbTarget.evict(targetID)
}
// Rename implements Archives. It renames every one of a database
// target's archive files to ArchiveFileName(webhookName, targetName,
// targetID), each keeping the period in its name, under the lock the
// target's archive writes and the idle sweep take. It never replaces
// a file: if one already has a new name, the error is
// ErrArchiveNameTaken. The caller renames before it saves the new
// name: see databaseTarget.rename.
func (e *Engine) Rename(
targetID, webhookName, targetName string,
) error {
if e.dbTarget == nil {
return nil
}
return e.dbTarget.rename(targetID, webhookName, targetName)
}
// StateAndCooldown implements CircuitBreakers. It returns the state of
// the target's circuit breaker and, while the breaker is open, what is
// left of its cooldown; the cooldown is zero once that has passed and
@@ -466,7 +409,7 @@ func (e *Engine) stop(ctx context.Context) error {
return err
}
e.dbTarget.evictAll()
e.archives.EvictAll()
e.log.Info("delivery engine stopped")
+11 -9
View File
@@ -427,9 +427,9 @@ func TestEngine_StopLeavesLastRetryAttemptRetrying(t *testing.T) {
// deliverToArchive gives the setup's webhook a database target,
// runs one delivery to it through the running engine, and returns
// the target's ID and archive file path. The archive writer holds
// the file open afterwards.
func deliverToArchive(t *testing.T, s iSetup) (string, string) {
// the target's archive file path. The archive writer holds the file
// open afterwards.
func deliverToArchive(t *testing.T, s iSetup) string {
t.Helper()
iCreateWebhook(t, s.MainDB, s.WebhookID, "hook")
@@ -452,7 +452,7 @@ func deliverToArchive(t *testing.T, s iSetup) (string, string) {
iWaitForDelivered(t, s.WebhookDB, deliveryID)
return tgt.ID, filepath.Join(
return filepath.Join(
filepath.Dir(s.DBMgr.DBPath(s.WebhookID)),
"archive-hook-archive-"+tgt.ID+".db",
)
@@ -471,7 +471,7 @@ func TestEngine_StopHookClosesArchives(t *testing.T) {
lc := startEngineViaHook(t, s.Engine)
_, path := deliverToArchive(t, s)
path := deliverToArchive(t, s)
require.FileExists(
t, path+"-wal",
"an open archive should have a -wal for the stop to remove",
@@ -505,21 +505,23 @@ func TestEngine_StopHookTimeoutLeavesArchivesOpen(t *testing.T) {
lc := startEngineViaHook(t, s.Engine)
targetID, _ := deliverToArchive(t, s)
path := deliverToArchive(t, s)
release := make(chan struct{})
t.Cleanup(func() {
close(release)
s.Engine.EvictWebhook(s.WebhookID)
s.Engine.ExportArchives().EvictWebhook(s.WebhookID)
})
s.Engine.ExportWedgeWorker(release)
requireStopHookExpires(t, lc.hooks[0], "delivery engine")
require.True(
t, s.Engine.ExportArchiveHandleOpen(targetID),
// Closing the last handle on an archive removes its -wal, so a
// -wal still there is a writer still open.
require.FileExists(
t, path+"-wal",
"a stop that timed out must not close archive writers",
)
}
+20 -281
View File
@@ -12,16 +12,11 @@ import (
"github.com/prometheus/client_golang/prometheus"
"go.uber.org/fx"
"gorm.io/gorm"
"sneak.berlin/go/webhooker/internal/archive"
"sneak.berlin/go/webhooker/internal/database"
"sneak.berlin/go/webhooker/internal/metrics"
)
// 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
@@ -399,7 +394,8 @@ func (e *Engine) ExportRetryCh() chan Task {
}
// NewTestEngine creates an Engine for unit tests without
// database dependencies.
// database dependencies. Its archive writers have no data
// directory, so a delivery to a database target fails.
func NewTestEngine(
log *slog.Logger,
client *http.Client,
@@ -411,6 +407,7 @@ func NewTestEngine(
retryCh: make(chan Task, retryChannelSize),
workers: workers,
mtr: metrics.New(prometheus.NewRegistry()),
archives: archive.NewWriters(archive.WritersParams{Logger: log}),
}
e.initTargets(client)
@@ -423,9 +420,10 @@ func NewTestEngineSmallRetry(
log *slog.Logger,
) *Engine {
e := &Engine{
log: log,
retryCh: make(chan Task, 1),
mtr: metrics.New(prometheus.NewRegistry()),
log: log,
retryCh: make(chan Task, 1),
mtr: metrics.New(prometheus.NewRegistry()),
archives: archive.NewWriters(archive.WritersParams{Logger: log}),
}
e.initTargets(nil)
@@ -433,7 +431,8 @@ func NewTestEngineSmallRetry(
}
// NewTestEngineWithDB creates an Engine with a real
// database and dbManager for integration tests.
// database and dbManager for integration tests, and archive
// writers of its own over the same two.
func NewTestEngineWithDB(
db *database.Database,
dbMgr *database.WebhookDBManager,
@@ -449,12 +448,22 @@ func NewTestEngineWithDB(
retryCh: make(chan Task, retryChannelSize),
workers: workers,
mtr: metrics.New(prometheus.NewRegistry()),
archives: archive.NewWriters(archive.WritersParams{
Database: db,
DBManager: dbMgr,
Logger: log,
}),
}
e.initTargets(client)
return e
}
// ExportArchives returns the engine's archive writers.
func (e *Engine) ExportArchives() *archive.Writers {
return e.archives
}
// ExportSetMetrics substitutes the engine's metric set, so a test can
// assert on collectors registered on a registry it holds.
func (e *Engine) ExportSetMetrics(mtr *metrics.Set) {
@@ -477,273 +486,3 @@ func NewTestCircuitBreaker(
cooldown: cooldown,
}
}
// ExportArchivedEvent aliases the archive row type so black-box
// tests can construct and read archive rows.
type ExportArchivedEvent = archivedEvent
// ExportArchiveWriter wraps an archiveWriter so black-box tests
// can exercise the archive file mechanics.
type ExportArchiveWriter struct {
w *archiveWriter
}
// NewExportArchiveWriter builds an archive writer for tests,
// optionally overriding the reopen debounce (a non-positive
// debounce keeps the production default).
func NewExportArchiveWriter(
path string, log *slog.Logger, debounce time.Duration,
) *ExportArchiveWriter {
w := newArchiveWriter(path, log)
if debounce > 0 {
w.debounce = debounce
}
return &ExportArchiveWriter{w: w}
}
// Write archives a row through the writer, into the file named
// without a period.
func (e *ExportArchiveWriter) Write(
row ExportArchivedEvent, expiry time.Duration,
) error {
return e.w.write(row, expiry, "")
}
// WritePeriod archives a row through the writer, into the file for
// period.
func (e *ExportArchiveWriter) WritePeriod(
row ExportArchivedEvent, expiry time.Duration, period string,
) error {
return e.w.write(row, expiry, period)
}
// Open opens the archive file, pruning when expiry is positive.
func (e *ExportArchiveWriter) Open(expiry time.Duration) error {
return e.w.open(e.w.path, expiry)
}
// Reopen closes and reopens the archive file.
func (e *ExportArchiveWriter) Reopen(
expiry time.Duration,
) error {
return e.w.reopen(e.w.path, expiry)
}
// SetNow replaces the clock the writer measures its reopen
// debounce on.
func (e *ExportArchiveWriter) SetNow(now func() time.Time) {
e.w.now = now
}
// Reopens reports how many times the file has been opened.
func (e *ExportArchiveWriter) Reopens() int {
return e.w.reopens
}
// DB returns the writer's current open handle for row
// inspection in tests.
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(e.w.path, 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()
}
// Rename gives the archive file a new name in the same directory,
// as a rename of the webhook or target does.
func (e *ExportArchiveWriter) Rename(name string) error {
return e.w.rename(name)
}
// 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
}
// Same reports whether both wrappers refer to the very same
// underlying archive writer, so a test can prove a registry entry
// is the writer it was handed rather than a replacement.
func (e *ExportArchiveWriter) Same(
other *ExportArchiveWriter,
) bool {
return other != nil && e.w == other.w
}
// ExportArchiveWriterFor returns the archive writer the registry
// currently caches for a database target, 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(
targetID string,
) *ExportArchiveWriter {
e.dbTarget.mu.Lock()
defer e.dbTarget.mu.Unlock()
w, ok := e.dbTarget.writers[targetID]
if !ok {
return nil
}
return &ExportArchiveWriter{w: w}
}
// ExportHasArchiveWriter reports whether the database target
// type currently caches an archive writer for a target.
func (e *Engine) ExportHasArchiveWriter(
targetID string,
) bool {
e.dbTarget.mu.Lock()
defer e.dbTarget.mu.Unlock()
_, ok := e.dbTarget.writers[targetID]
return ok
}
// ExportArchiveHandleOpen reports whether the cached archive
// writer for a target holds an open database handle. It
// returns false when no writer is cached.
func (e *Engine) ExportArchiveHandleOpen(
targetID string,
) bool {
e.dbTarget.mu.Lock()
w, ok := e.dbTarget.writers[targetID]
e.dbTarget.mu.Unlock()
if !ok {
return false
}
w.mu.Lock()
defer w.mu.Unlock()
return w.db != nil
}
// ExportEnsureArchiveWriter creates (if needed) and returns the
// archive file path of the cached writer for a target, so a
// test can prime the registry the way a delivery would.
func (e *Engine) ExportEnsureArchiveWriter(
targetID string,
) (string, error) {
w, err := e.dbTarget.writerFor(targetID)
if err != nil {
return "", err
}
return w.path, nil
}
// ExportSweepWriterFor takes a target's registry writer exactly
// as the idle sweep does, reporting whether the sweep had to
// create the entry. It lets a test drive the registry through the
// sweep's own entry point instead of choreographing goroutines.
func (e *Engine) ExportSweepWriterFor(
targetID string,
) (*ExportArchiveWriter, bool, error) {
w, created, err := e.dbTarget.sweepWriterFor(targetID)
if err != nil {
return nil, false, err
}
return &ExportArchiveWriter{w: w}, created, nil
}
// ExportReleaseSweepWriter releases a sweep-created registry entry
// exactly as a finished sweep does.
func (e *Engine) ExportReleaseSweepWriter(
targetID string, w *ExportArchiveWriter,
) {
e.dbTarget.releaseSweepWriter(targetID, w.w)
}
// NewTestArchiveSweeper builds an ArchiveSweeper backed by the
// given main database and engine, without the fx lifecycle.
// Intended for tests.
func NewTestArchiveSweeper(
db *database.Database,
eng *Engine,
log *slog.Logger,
) *ArchiveSweeper {
return &ArchiveSweeper{
db: db,
eng: eng,
log: log,
interval: time.Hour,
}
}
// ExportSweep runs a single archive sweep synchronously for
// tests.
func (s *ArchiveSweeper) ExportSweep(ctx context.Context) {
s.sweep(ctx)
}
// 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)
}
// ExportStop stops the sweeper's background loop for tests.
func (s *ArchiveSweeper) ExportStop(ctx context.Context) error {
return s.stop(ctx)
}
// ExportWedgeLoop adds a goroutine to the sweeper's WaitGroup
// that never observes cancellation and returns only when release
// is closed. It stands in for a prune stuck on a locked archive.
func (s *ArchiveSweeper) ExportWedgeLoop(
release <-chan struct{},
) {
s.wg.Go(func() {
<-release
})
}
// ExportSetInterval overrides the sweep interval for tests.
func (s *ArchiveSweeper) ExportSetInterval(d time.Duration) {
s.interval = d
}
// ExportParseArchiveExpiry exposes parseArchiveExpiry.
func ExportParseArchiveExpiry(
configJSON string,
) (time.Duration, error) {
return parseArchiveExpiry(configJSON)
}
-29
View File
@@ -2,8 +2,6 @@ package delivery_test
import (
"context"
"path/filepath"
"strings"
"sync"
"testing"
"time"
@@ -66,33 +64,6 @@ func fDrain(e *delivery.Engine) []delivery.Task {
}
}
// 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
+1 -4
View File
@@ -103,16 +103,13 @@ func (e *Engine) initTargets(client *http.Client) {
client: client,
}
dbT := &databaseTarget{eng: e}
e.httpTarget = httpT
e.slackTarget = slackT
e.dbTarget = dbT
e.targets = map[database.TargetType]Target{
database.TargetTypeHTTP: httpT,
database.TargetTypeSlack: slackT,
database.TargetTypeDatabase: dbT,
database.TargetTypeDatabase: &databaseTarget{eng: e},
database.TargetTypeLog: &logTarget{eng: e},
}
}
+5 -4
View File
@@ -5,6 +5,7 @@ import (
"errors"
"fmt"
"sneak.berlin/go/webhooker/internal/archive"
"sneak.berlin/go/webhooker/internal/database"
)
@@ -105,7 +106,7 @@ func databaseConfigForm(
return TargetConfigForm{}, nil
}
var cfg databaseTargetConfig
var cfg archive.DatabaseTargetConfig
err := json.Unmarshal([]byte(configJSON), &cfg)
if err != nil {
@@ -114,18 +115,18 @@ func databaseConfigForm(
)
}
err = ValidateArchiveRotation(cfg.Rotation)
err = archive.ValidateRotation(cfg.Rotation)
if err != nil {
return TargetConfigForm{}, err
}
form := TargetConfigForm{Rotation: cfg.Rotation}
if cfg.Expiry == "" || cfg.Expiry == archiveExpiryNever {
if cfg.Expiry == "" || cfg.Expiry == archive.ExpiryNever {
return form, nil
}
err = ValidateArchiveExpiry(cfg.Expiry)
err = archive.ValidateExpiry(cfg.Expiry)
if err != nil {
return TargetConfigForm{}, err
}
+4 -3
View File
@@ -5,6 +5,7 @@ import (
"strconv"
"time"
"sneak.berlin/go/webhooker/internal/archive"
"sneak.berlin/go/webhooker/internal/database"
)
@@ -200,17 +201,17 @@ func maxAttemptsField(t *database.Target) ConfigField {
// not one of the four, is reported as unavailable rather than
// echoed back.
func databaseConfigFields(configJSON string) []ConfigField {
expiry, err := parseArchiveExpiry(configJSON)
expiry, err := archive.ParseExpiry(configJSON)
if err != nil {
return unavailableConfigFields()
}
rotation, err := parseArchiveRotation(configJSON)
rotation, err := archive.ParseRotation(configJSON)
if err != nil {
return unavailableConfigFields()
}
value := archiveExpiryNever
value := archive.ExpiryNever
if expiry > 0 {
value = plainDuration(expiry)
}
@@ -35,6 +35,17 @@ const (
viewMaxAttempts = "Delivery attempts"
)
// The archive rotations a database target's settings can name, and
// the config of one whose rotation is not one of the four.
const (
rotationNone = "none"
rotationMonthly = "monthly"
rotationDaily = "daily"
rotationHourly = "hourly"
weeklyConfig = `{"rotation":"weekly"}`
)
func TestMaskedWebhookURL(t *testing.T) {
t.Parallel()
+4 -369
View File
@@ -2,83 +2,23 @@ package delivery
import (
"context"
"fmt"
"strings"
"sync"
"time"
"gorm.io/gorm"
"sneak.berlin/go/webhooker/internal/database"
)
// archiveNameMaxLen is how many characters of a webhook or target
// name an archive file name keeps.
const archiveNameMaxLen = 40
// databaseTarget is a no-retry target that archives the full
// inbound event into the target's own archive SQLite file, separate
// from the per-webhook event database. The event is already
// persisted in the per-webhook event DB by the time delivery runs;
// the database target additionally writes a durable long-term copy
// into the file ArchiveFileName names, with a period added when the
// target rotates (see archivePeriodPath), and then records a single
// through the engine's archive.Writers, and then records a single
// attempt whose outcome reflects whether the archive write
// succeeded. See archiveWriter for the close/reopen, auto-recreate,
// and expiry semantics.
// succeeded. Package archive holds everything about the files
// themselves.
type databaseTarget struct {
eng *Engine
// writers holds one archive writer per database target, keyed
// by target ID.
mu sync.Mutex
writers map[string]*archiveWriter
}
// ArchiveFileName returns the file name of a database target's
// archive: archive-WEBHOOKNAME-TARGETNAME-TARGETID.db, with both
// names passed through archiveNamePart. The target ID keeps the
// name unique when two targets' names come out the same.
func ArchiveFileName(webhookName, targetName, targetID string) string {
return "archive-" + archiveNamePart(webhookName) + "-" +
archiveNamePart(targetName) + "-" + targetID + ".db"
}
// archiveNamePart makes a webhook or target name safe to put in a
// file name. It is lowercased; ASCII letters and digits are kept,
// every other run of characters becomes a single "-", and no "-" is
// left at either end. It is cut to archiveNameMaxLen characters, and
// a name with nothing left is "unnamed".
func archiveNamePart(name string) string {
var b strings.Builder
dash := false
for _, r := range strings.ToLower(name) {
if (r < 'a' || r > 'z') && (r < '0' || r > '9') {
dash = b.Len() > 0
continue
}
if dash {
b.WriteByte('-')
dash = false
}
b.WriteRune(r)
}
part := b.String()
if len(part) > archiveNameMaxLen {
part = strings.TrimRight(part[:archiveNameMaxLen], "-")
}
if part == "" {
return "unnamed"
}
return part
}
// Deliver implements Target. It archives the event, then
@@ -98,7 +38,7 @@ func (t *databaseTarget) Deliver(
) {
start := time.Now()
err := t.archive(d)
err := t.eng.archives.Write(d)
elapsed := time.Since(start)
@@ -145,308 +85,3 @@ func (t *databaseTarget) Deliver(
database.DeliveryStatusDelivered,
)
}
// archive writes the full event as a row into the target's
// archive database, honouring the optional per-target expiry and
// rotation parsed from the target config JSON. With rotation, the
// event goes to the file for the period of its receive time.
func (t *databaseTarget) archive(d *database.Delivery) error {
webhookID := d.Event.WebhookID
if webhookID == "" {
return errArchiveMissingWebhookID
}
expiry, err := parseArchiveExpiry(d.Target.Config)
if err != nil {
return err
}
rotation, err := parseArchiveRotation(d.Target.Config)
if err != nil {
return err
}
// An event whose stored row was gone before its delivery ran has
// no receive time (see Engine.hydrateEvent), and goes to the
// file for now.
receivedAt := d.Event.CreatedAt
if receivedAt.IsZero() {
receivedAt = time.Now()
}
w, err := t.writerFor(d.TargetID)
if err != nil {
return err
}
row := archivedEvent{
EventID: d.Event.ID,
WebhookID: webhookID,
EntrypointID: d.Event.EntrypointID,
Method: d.Event.Method,
RawQuery: d.Event.RawQuery,
Headers: d.Event.Headers,
Body: d.Event.Body,
ContentType: d.Event.ContentType,
}
return w.write(row, expiry, archivePeriod(rotation, receivedAt))
}
// writerFor returns the archive writer for a database target,
// creating and caching it on first use. Each target has one writer
// so its close/reopen debounce state is shared across concurrent
// deliveries, and so a rename and the idle sweep take the same lock
// as its writes.
func (t *databaseTarget) writerFor(
targetID string,
) (*archiveWriter, error) {
t.mu.Lock()
defer t.mu.Unlock()
w, ok := t.writers[targetID]
if !ok {
var err error
w, err = t.newWriter(targetID)
if err != nil {
return nil, err
}
if t.writers == nil {
t.writers = make(map[string]*archiveWriter)
}
t.writers[targetID] = 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 target's archive 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 target'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(
targetID string,
) (*archiveWriter, bool, error) {
t.mu.Lock()
defer t.mu.Unlock()
w, ok := t.writers[targetID]
if ok {
return w, false, nil
}
w, err := t.newWriter(targetID)
if err != nil {
return nil, false, err
}
if t.writers == nil {
t.writers = make(map[string]*archiveWriter)
}
w.sweepOwned = true
t.writers[targetID] = 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(
targetID string, w *archiveWriter,
) {
t.mu.Lock()
defer t.mu.Unlock()
cur, ok := t.writers[targetID]
if !ok || cur != w || !cur.sweepOwned {
return
}
delete(t.writers, targetID)
}
// newWriter builds the writer for a database target's archive. Its
// path is the one ArchivePath gives for the webhook and the target as
// the main database names them now; from then on only rename changes
// the name the writer uses. It does not touch the archive files.
func (t *databaseTarget) newWriter(
targetID string,
) (*archiveWriter, error) {
if t.eng.dbManager == nil {
return nil, errArchiveNoDataDir
}
var target database.Target
err := t.eng.database.DB().
Preload("Webhook").
First(&target, "id = ?", targetID).Error
if err != nil {
return nil, fmt.Errorf(
"loading database target %s: %w", targetID, err,
)
}
w := newArchiveWriter(
ArchivePath(t.eng.dbManager, &target.Webhook, &target),
t.eng.log,
)
w.webhookID = target.WebhookID
return w, nil
}
// rename moves every one of a database target's archive files to the
// name for webhookName and targetName. It goes through the target's
// writer, so the move holds the lock that writes and the idle sweep
// take, and later writes use the new name.
//
// The writer is created if there is none, and it stays cached. The
// handlers rename before they save the new name, so until the save
// the main database still has the old one; a delivery in that window
// must find this writer rather than build one from the old name.
func (t *databaseTarget) rename(
targetID, webhookName, targetName string,
) error {
w, err := t.writerFor(targetID)
if err != nil {
return err
}
return w.rename(ArchiveFileName(webhookName, targetName, targetID))
}
// evict drops a database target's archive writer from the registry
// and closes its handle, so a deleted target does not leave a
// writer (and an open archive handle within its debounce window)
// alive for the process lifetime.
//
// The map entry is removed under the registry lock, which is
// then released before the handle is closed under the writer's
// own lock: that ordering keeps the registry available to other
// targets while an in-flight write on this one drains, and
// closing under the writer's lock means eviction can never race
// a write.
//
// Eviction is idempotent and silent for a target with no writer,
// which is the common case: only a database target that has
// received an event or been renamed has one. It never deletes the
// archive file.
func (t *databaseTarget) evict(targetID string) {
t.mu.Lock()
w, ok := t.writers[targetID]
if ok {
delete(t.writers, targetID)
}
t.mu.Unlock()
if !ok {
return
}
w.evict()
t.eng.log.Info(
"evicted archive writer",
"target_id", targetID,
"path", w.path,
)
}
// evictWebhook evicts, exactly as evict does, the writer of every
// database target of a webhook.
func (t *databaseTarget) evictWebhook(webhookID string) {
t.mu.Lock()
var gone []*archiveWriter
for targetID, w := range t.writers {
if w.webhookID == webhookID {
delete(t.writers, targetID)
gone = append(gone, w)
}
}
t.mu.Unlock()
for _, w := range gone {
w.evict()
t.eng.log.Info(
"evicted archive writer",
"webhook_id", webhookID,
"path", w.path,
)
}
}
// evictAll evicts every cached archive writer, exactly as evict
// does for one target. 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()
}
}
// sweepArchive prunes one database target's archive of rows older
// than expiry, without requiring a write. A missing archive file is
// left missing (see sweepExpired), so a sweep never creates an
// archive for a target that 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 target deletion cannot
// resurrect the writer the eviction just dropped.
func (t *databaseTarget) sweepArchive(
targetID string, expiry time.Duration,
) error {
w, created, err := t.sweepWriterFor(targetID)
if err != nil {
return err
}
if created {
defer t.releaseSweepWriter(targetID, w)
}
return w.sweepExpired(expiry)
}
@@ -1,694 +0,0 @@
package delivery
import (
"encoding/json"
"errors"
"fmt"
"io/fs"
"log/slog"
"os"
"path/filepath"
"sync"
"time"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
"sneak.berlin/go/webhooker/internal/database"
"sneak.berlin/go/webhooker/internal/db/migrations"
"sneak.berlin/go/webhooker/internal/gormlog"
)
// archiveExpiryNever is the expiry sentinel (and default) that
// disables pruning so archived rows are kept forever.
const archiveExpiryNever = "never"
// archiveReopenDebounce bounds how often an archive file is
// closed and reopened. After each write the handle is closed
// and reopened so an operator can move the file away for
// offline archiving, but never more than once per this window.
const archiveReopenDebounce = time.Second
const (
// archiveModeCreate is the SQLite URI mode used by the write
// 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 = 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 = database.SQLiteModeExisting
)
var (
// errArchiveMissingWebhookID is returned when an event to
// archive has no webhook id to record in its archive row.
errArchiveMissingWebhookID = errors.New(
"cannot archive event without a webhook id",
)
// errArchiveNoDataDir is returned when the database target
// has no webhook database manager and so cannot locate the
// data directory for archive files.
errArchiveNoDataDir = errors.New(
"database target has no data directory",
)
// errArchiveExpiryNotPositive is returned when a
// user-supplied archive expiry parses as a duration but is
// zero or negative; "never" is the way to disable pruning.
errArchiveExpiryNotPositive = errors.New(
"expiry must be a positive duration or \"never\"",
)
// errArchiveWriterEvicted is returned when a writer that has
// been evicted (its target or its webhook was deleted) is used
// again. An evicted writer is no longer in the registry, so
// reopening its file would leak a handle nothing owns.
errArchiveWriterEvicted = errors.New(
"archive writer has been evicted",
)
// ErrArchiveNameTaken is returned when an archive cannot be
// renamed because a file already has the new name. That file may
// be an archive with rows of its own, so it is never replaced.
ErrArchiveNameTaken = errors.New(
"a file already has the archive's new name",
)
)
// databaseTargetConfig is the optional per-target JSON config
// for a database (archive) target.
type databaseTargetConfig struct {
// Expiry is a Go duration (e.g. "720h") after which
// archived rows are pruned, or "never" (the default) to
// keep them forever.
Expiry string `json:"expiry"`
// Rotation is none (the default), monthly, daily or hourly: see
// archivePeriod.
Rotation string `json:"rotation"`
}
// archivedEvent is one fully captured webhook event stored in a
// database target's archive for long-term retention. It is a
// self-contained copy — independent of the per-webhook event
// database, which may prune events under its own retention.
type archivedEvent struct {
ID uint `gorm:"primaryKey;autoIncrement"`
EventID string `gorm:"index"`
WebhookID string
EntrypointID string
Method string
RawQuery string
Headers string
Body string
ContentType string
// ArchivedAt is when the row was archived and is the age
// basis for expiry pruning.
ArchivedAt time.Time `gorm:"index"`
}
// parseArchiveExpiry reads the optional expiry from a database
// target's config JSON. An empty config, an empty expiry, or
// the literal "never" all mean keep forever, returned as a zero
// duration. Any other value must parse as a positive Go
// duration; a set-but-invalid value (unparseable, zero, or
// negative) is an error rather than a silent default, matching
// ValidateArchiveExpiry at target creation.
func parseArchiveExpiry(
configJSON string,
) (time.Duration, error) {
if configJSON == "" {
return 0, nil
}
var cfg databaseTargetConfig
err := json.Unmarshal([]byte(configJSON), &cfg)
if err != nil {
return 0, fmt.Errorf(
"parsing database target config: %w", err,
)
}
if cfg.Expiry == "" || cfg.Expiry == archiveExpiryNever {
return 0, nil
}
dur, err := time.ParseDuration(cfg.Expiry)
if err != nil {
return 0, fmt.Errorf(
"parsing archive expiry %q: %w", cfg.Expiry, err,
)
}
if dur <= 0 {
return 0, fmt.Errorf(
"%w: %q", errArchiveExpiryNotPositive, cfg.Expiry,
)
}
return dur, nil
}
// ValidateArchiveExpiry checks a user-supplied archive expiry
// for a database target at configuration time. Valid values are
// empty, "never" (both meaning keep forever), or a positive Go
// duration such as "720h". Anything else is an error, so a bad
// expiry is rejected when the target is created rather than
// failing every subsequent delivery.
func ValidateArchiveExpiry(expiry string) error {
if expiry == "" || expiry == archiveExpiryNever {
return nil
}
dur, err := time.ParseDuration(expiry)
if err != nil {
return fmt.Errorf(
"expiry must be %q or a Go duration "+
"such as \"720h\": %w",
archiveExpiryNever, err,
)
}
if dur <= 0 {
return fmt.Errorf(
"%w: %q", errArchiveExpiryNotPositive, expiry,
)
}
return nil
}
// archiveWriter owns one database target's archive SQLite files.
// It serialises writes, and after each write closes and reopens
// the file (debounced to at most once per debounce window) so
// an operator can move the file away for offline archiving. The
// next write recreates a moved or removed file, because the
// file is opened create-if-missing and its schema is migrated
// on every open.
type archiveWriter struct {
mu sync.Mutex
// path is the target's archive file as ArchivePath names it. A
// target that rotates writes to the files archivePeriodPath names
// for path and a period instead.
path string
// current is the file db is open on.
current string
log *slog.Logger
debounce time.Duration
db *gorm.DB
lastReopen time.Time
reopens int
// now is the clock the reopen debounce is measured on. It is
// time.Now outside tests.
now func() time.Time
// evicted marks a writer that has been removed from the
// registry. Its handle is closed and it must never open the
// file again: nothing holds it any more, so a reopen would
// leak the handle for the process lifetime.
evicted bool
// webhookID is the webhook the archive's target belongs to,
// so deleting the webhook can find its writers. It is set
// when the writer is created and never changes.
webhookID string
// sweepOwned marks a registry entry that the idle sweep
// created because no writer was cached for the target. The
// sweep removes such an entry again when it is done, so a
// sweep can never leave — or resurrect — a registry entry
// for a target 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
// the default reopen debounce.
func newArchiveWriter(
path string, log *slog.Logger,
) *archiveWriter {
return &archiveWriter{
path: path,
log: log,
debounce: archiveReopenDebounce,
now: time.Now,
}
}
// write appends the event as a row to the archive file for period
// (see archivePeriodPath), then applies the debounced close/reopen.
// When period names a different file from the one open, the open one
// is closed first. It recreates the archive file if it was moved or
// removed since the last open. A positive expiry prunes rows older
// than it on each (re)open.
func (w *archiveWriter) write(
row archivedEvent, expiry time.Duration, period string,
) error {
w.mu.Lock()
defer w.mu.Unlock()
if w.evicted {
return fmt.Errorf(
"%w: %s", errArchiveWriterEvicted, w.path,
)
}
file := archivePeriodPath(w.path, period)
if w.db == nil || w.current != file || !fileExists(file) {
err := w.reopen(file, expiry)
if err != nil {
return err
}
}
// GORM's clock, the one that stamps created_at in the other
// databases, so this is stored in the same form. A bare time.Now()
// carries a monotonic clock reading, which the driver would store
// as an " m=+…" suffix.
row.ArchivedAt = w.db.NowFunc()
err := w.db.Create(&row).Error
if err != nil {
return fmt.Errorf(
"archiving event to %s: %w", file, err,
)
}
if w.now().Sub(w.lastReopen) >= w.debounce {
return w.reopen(file, expiry)
}
return nil
}
// open opens (creating if missing) an archive file, migrates
// its schema, records the reopen time, and prunes expired rows
// when expiry is positive.
func (w *archiveWriter) open(file string, expiry time.Duration) error {
return w.openMode(file, archiveModeCreate, expiry)
}
// openMode opens an archive file with the given SQLite URI
// mode, migrates its schema, records the reopen time, and
// prunes expired rows when expiry is positive. The write path
// passes archiveModeCreate so a missing file is recreated; the
// idle sweep passes archiveModeExisting so a missing file is an
// error rather than a newly conjured empty archive.
func (w *archiveWriter) openMode(
file, mode string, expiry time.Duration,
) error {
// 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(file, mode)
if err != nil {
return fmt.Errorf(
"opening archive database %s: %w", file, err,
)
}
gdb, err := gorm.Open(
sqlite.Dialector{Conn: sqlDB}, &gorm.Config{
// Never leave this at GORM's default. See
// internal/gormlog.
Logger: gormlog.New(w.log),
},
)
if err != nil {
_ = sqlDB.Close()
return fmt.Errorf(
"connecting to archive database %s: %w",
file, err,
)
}
err = migrations.Apply(gdb, migrations.Archive)
if err != nil {
_ = sqlDB.Close()
return fmt.Errorf(
"migrating archive database %s: %w", file, err,
)
}
w.db = gdb
w.current = file
w.lastReopen = w.now()
w.reopens++
if expiry > 0 {
w.prune(expiry)
}
return nil
}
// reopen closes any open handle and opens file afresh. The
// fresh open recreates the file if it was moved away.
func (w *archiveWriter) reopen(file string, expiry time.Duration) error {
w.close()
return w.open(file, expiry)
}
// close closes the underlying handle, if any.
func (w *archiveWriter) close() {
if w.db == nil {
return
}
sqlDB, err := w.db.DB()
if err == nil {
_ = sqlDB.Close()
}
w.db = nil
}
// sweepExpired prunes the target's archive files, which may have
// gone idle, with no write to trigger the usual on-reopen prune. It
// lists the files under the writer's own mutex, then takes the mutex
// again for one file at a time, so a write waits for at most one
// file's prune, and each prune is ordered against concurrent writes
// rather than reaching around them to the file.
//
// It never creates an archive file: it prunes only the files
// archiveFiles lists, skips one that is gone by the time it is
// reached (moved away, or renamed since the listing), and opens each
// with archiveModeExisting so SQLite itself refuses to create one if
// the file disappears between the check and the open. A file named
// for a period that the prune leaves empty is deleted.
//
// The archive is left CLOSED afterwards. An idle archive holding
// no handle is what keeps the operator's move-the-file-away
// workflow working; the next write reopens (and recreates) the
// file as it always has.
func (w *archiveWriter) sweepExpired(expiry time.Duration) error {
w.mu.Lock()
files, err := archiveFiles(w.path)
w.mu.Unlock()
if err != nil {
return err
}
var errs []error
for _, file := range files {
err = w.sweepFile(file, expiry)
if errors.Is(err, errArchiveWriterEvicted) {
return err
}
if err != nil {
errs = append(errs, err)
}
}
return errors.Join(errs...)
}
// sweepFile prunes one of the target's archive files for sweepExpired,
// holding w.mu while it does. It skips a file that is gone, and deletes
// the file, with its -wal and -shm, when it is named for a period and
// the prune leaves it empty.
func (w *archiveWriter) sweepFile(
file archiveFile, expiry time.Duration,
) error {
w.mu.Lock()
defer w.mu.Unlock()
if w.evicted {
return fmt.Errorf(
"%w: %s", errArchiveWriterEvicted, w.path,
)
}
if !fileExists(file.path) {
return nil
}
// Drop any live handle first so the prune runs against a
// freshly opened file, matching the write path's semantics.
w.close()
err := w.openMode(file.path, archiveModeExisting, expiry)
if err != nil {
return err
}
if file.period == "" {
w.close()
return nil
}
var rows int64
err = w.db.Model(&archivedEvent{}).Count(&rows).Error
w.close()
if err != nil {
return fmt.Errorf(
"counting rows in archive %s: %w", file.path, err,
)
}
if rows > 0 {
return nil
}
for _, suffix := range []string{"", "-wal", "-shm"} {
err = os.Remove(file.path + suffix)
if err != nil && !errors.Is(err, fs.ErrNotExist) {
return fmt.Errorf("deleting empty archive file: %w", err)
}
}
w.log.Info("deleted empty archive file", "path", file.path)
return nil
}
// rename gives every one of the target's archive files the new
// name, keeping the period in the name of each (see
// archivePeriodPath), and the writer uses the files under that name
// from now on. The handle is closed first, which folds the -wal into
// the .db; any -wal or -shm still beside a file (left by a crash) is
// moved with it, because SQLite finds them by name. A target with no
// files is not an error: the operator may have moved them away, and
// the next write creates its file under the new name.
//
// If a file already has one of the new names, nothing is moved and
// the error is ErrArchiveNameTaken. If one file fails to move, those
// already moved are moved back before the error is returned, so the
// archive is never split across two names.
func (w *archiveWriter) rename(name string) error {
w.mu.Lock()
defer w.mu.Unlock()
if w.evicted {
return fmt.Errorf(
"%w: %s", errArchiveWriterEvicted, w.path,
)
}
path := filepath.Join(filepath.Dir(w.path), name)
if path == w.path {
return nil
}
files, err := archiveFiles(w.path)
if err != nil {
return err
}
// from[i] moves to to[i].
var from, to []string
for _, file := range files {
renamed := archivePeriodPath(path, file.period)
for _, suffix := range []string{"", "-wal", "-shm"} {
from = append(from, file.path+suffix)
to = append(to, renamed+suffix)
}
}
for _, taken := range to {
if fileExists(taken) {
return fmt.Errorf(
"%w: %s", ErrArchiveNameTaken, filepath.Base(taken),
)
}
}
w.close()
for i := range from {
err = os.Rename(from[i], to[i])
if err == nil || errors.Is(err, fs.ErrNotExist) {
continue
}
for j := range i {
backErr := os.Rename(to[j], from[j])
if backErr != nil && !errors.Is(backErr, fs.ErrNotExist) {
w.log.Error(
"failed to move archive file back",
"from", to[j],
"to", from[j],
"error", backErr,
)
}
}
return fmt.Errorf(
"renaming archive %s to %s: %w", from[i], to[i], err,
)
}
w.path = path
return nil
}
// evict closes the writer's handle and marks it unusable. It is
// called when the writer leaves the registry, because its target
// or its webhook was deleted, or at shutdown. The archive FILE is
// deliberately left on disk: it is long-term storage an operator
// may still want.
func (w *archiveWriter) evict() {
w.mu.Lock()
defer w.mu.Unlock()
w.evicted = true
w.close()
}
// prune deletes archived rows older than expiry, measured from
// each row's archived time. It runs on every (re)open, so a
// steadily written archive is swept by its own write traffic. An
// archive that goes idle receives no further reopens, which is
// why ArchiveSweeper exists to drive sweepExpired on a timer.
// Failures are logged, not fatal: a prune error must not stop
// archiving.
func (w *archiveWriter) prune(expiry time.Duration) {
cutoff := time.Now().Add(-expiry)
res := w.db.Where("archived_at < ?", cutoff).
Delete(&archivedEvent{})
if res.Error != nil {
w.log.Error(
"failed to prune expired archive rows",
"path", w.current,
"error", res.Error,
)
return
}
if res.RowsAffected > 0 {
w.log.Info(
"pruned expired archive rows",
"path", w.current,
"rows_deleted", res.RowsAffected,
)
}
}
// ArchiveFileInfo is what the metadata of a database target's archive
// files says about them.
type ArchiveFileInfo struct {
// Files counts the files.
Files int
// Size is the bytes on disk of the files and their -wal together.
Size int64
// Written is when a file or a -wal was last modified, whichever is
// latest: a write lands in the -wal first.
Written time.Time
}
// StatArchive reads the metadata of a database target's archive
// files, given the path ArchivePath gives it (see archiveFiles), and
// of their -wal, without opening them. With no files, which is so
// before the first write and after the operator moved them away, the
// error wraps fs.ErrNotExist.
func StatArchive(path string) (ArchiveFileInfo, error) {
files, err := archiveFiles(path)
if err != nil {
return ArchiveFileInfo{}, err
}
var info ArchiveFileInfo
for _, file := range files {
db, err := os.Stat(file.path)
if errors.Is(err, fs.ErrNotExist) {
continue
}
if err != nil {
return ArchiveFileInfo{}, err
}
info.Files++
info.Size += db.Size()
if db.ModTime().After(info.Written) {
info.Written = db.ModTime()
}
wal, err := os.Stat(file.path + "-wal")
if errors.Is(err, fs.ErrNotExist) {
continue
}
if err != nil {
return ArchiveFileInfo{}, err
}
info.Size += wal.Size()
if wal.ModTime().After(info.Written) {
info.Written = wal.ModTime()
}
}
if info.Files == 0 {
return ArchiveFileInfo{}, fmt.Errorf(
"no archive file for %s: %w", path, fs.ErrNotExist,
)
}
return info, nil
}
// fileExists reports whether a path currently exists.
func fileExists(path string) bool {
_, err := os.Stat(path)
return err == nil
}
@@ -1,155 +0,0 @@
package delivery_test
import (
"bytes"
"log"
"log/slog"
"path/filepath"
"strings"
"sync"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"gorm.io/gorm"
gormlogger "gorm.io/gorm/logger"
"sneak.berlin/go/webhooker/internal/delivery"
"sneak.berlin/go/webhooker/internal/middleware"
)
// archiveGORMTailMarker sits at the far end of the value this file
// drives into an archive lookup. Its presence in a log line means the
// whole value reached the log, so nothing truncated it.
const archiveGORMTailMarker = "ENDOFCLIENTVALUE"
// archiveGORMFillBytes is how much text the lookup carries. It is far
// past every budget in play.
const archiveGORMFillBytes = 8 << 10
// gormDefaultBuf collects what GORM's package-level default logger
// writes, if anything reaches it.
type gormDefaultBuf struct {
mu sync.Mutex
b bytes.Buffer
}
func (g *gormDefaultBuf) Write(p []byte) (int, error) {
g.mu.Lock()
defer g.mu.Unlock()
return g.b.Write(p)
}
func (g *gormDefaultBuf) String() string {
g.mu.Lock()
defer g.mu.Unlock()
return g.b.String()
}
// captureArchiveGORMDefault replaces GORM's package-level default
// logger with one configured exactly as GORM configures its own,
// writing to a buffer.
//
// This duplicates the detector in internal/handlers rather than
// sharing it: a test helper cannot cross a package's test boundary
// without exporting production code to carry it, and a logging
// detector is not worth a production symbol. What it detects is the
// third gorm.Open in this service, at
// internal/delivery/target_database_archive.go — the archive writer,
// whose type is unexported, so nothing outside this package can drive
// it.
func captureArchiveGORMDefault(t *testing.T) *gormDefaultBuf {
t.Helper()
buf := &gormDefaultBuf{}
orig := gormlogger.Default
gormlogger.Default = gormlogger.New(
log.New(buf, "", log.LstdFlags),
gormlogger.Config{
SlowThreshold: 200 * time.Millisecond,
LogLevel: gormlogger.Warn,
IgnoreRecordNotFoundError: false,
Colorful: false,
},
)
t.Cleanup(func() { gormlogger.Default = orig })
return buf
}
// TestArchiveWriter_NeverUsesGORMsDefaultLogger pins the archive
// writer's gorm.Open to the adapter.
//
// Restore a bare &gorm.Config{} at
// internal/delivery/target_database_archive.go and this fails: the
// default logger prints the fully interpolated SELECT on every
// ErrRecordNotFound, so the client-chosen event id below arrives whole
// and unbounded on stdout, answering to no level the operator set.
//
// Not parallel: gormlogger.Default is process-global. Go runs every
// non-parallel top-level test to completion before it resumes the
// parallel ones.
//
//nolint:paralleltest // Deliberately sequential; see above.
func TestArchiveWriter_NeverUsesGORMsDefaultLogger(t *testing.T) {
var captured bytes.Buffer
gormDefault := captureArchiveGORMDefault(t)
w := delivery.NewExportArchiveWriter(
filepath.Join(t.TempDir(), "archive.db"),
slog.New(slog.NewTextHandler(
&captured, &slog.HandlerOptions{Level: slog.LevelDebug},
)),
0,
)
require.NoError(t, w.Open(0))
t.Cleanup(w.Evict)
// A lookup that misses, carrying a value the size of an inbound
// event id. Under the default logger this is the line that gets
// interpolated and printed.
value := strings.Repeat("\x01", archiveGORMFillBytes) +
archiveGORMTailMarker
var row delivery.ExportArchivedEvent
err := w.DB().Where("event_id = ?", value).First(&row).Error
require.ErrorIs(t, err, gorm.ErrRecordNotFound)
got := gormDefault.String()
assert.Empty(
t, got,
"GORM's default logger wrote %d bytes, so the archive "+
"writer's gorm.Open is back on a bare &gorm.Config{}; "+
"the first of them: %s",
len(got), got[:min(len(got), 300)],
)
// The adapter drops a miss, so this should be silent too — and
// whatever it does write stays inside the stated ceiling.
out := captured.String()
assert.NotContains(
t, out, archiveGORMTailMarker,
"the far end of the client-chosen value reached the log",
)
for line := range strings.SplitSeq(strings.TrimRight(out, "\n"), "\n") {
if line == "" {
continue
}
assert.LessOrEqual(
t, len(line), middleware.MaxAccessLogLineBytes,
"log line exceeded its bound: %s",
line[:min(len(line), 300)],
)
}
}
@@ -1,417 +0,0 @@
package delivery_test
import (
"context"
"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"
)
// deliverTo archives one event to a database target, leaving the
// target's writer cached with its handle open.
func deliverTo(
t *testing.T, env *archiveEnv, tgt *database.Target,
) {
t.Helper()
webhookDB := testWebhookDB(t)
event := seedEvent(t, webhookDB, `{"archived":true}`)
env.eng.ExportDeliverDatabase(
webhookDB, seedDatabaseTargetDelivery(t, webhookDB, event, tgt),
)
}
// TestEvictWebhook_ClosesAndRemovesWriter proves that evicting
// a webhook drops the archive writers of its database targets
// from the registry and closes their open handles, rather than
// leaving them alive for the process lifetime, and leaves another
// webhook's writer alone.
func TestEvictWebhook_ClosesAndRemovesWriter(t *testing.T) {
t.Parallel()
env := setupArchiveTest(t)
first := env.seedDatabaseTarget(t, "")
second := env.addDatabaseTarget(t, first.WebhookID, "")
other := env.seedDatabaseTarget(t, "")
for _, tgt := range []*database.Target{first, second, other} {
deliverTo(t, env, tgt)
require.True(
t, env.eng.ExportArchiveHandleOpen(tgt.ID),
"the writer should hold an open handle after a write",
)
}
env.eng.EvictWebhook(first.WebhookID)
for _, tgt := range []*database.Target{first, second} {
assert.False(
t, env.eng.ExportHasArchiveWriter(tgt.ID),
"eviction should remove the registry entry",
)
assert.False(
t, env.eng.ExportArchiveHandleOpen(tgt.ID),
"eviction should close the archive handle",
)
assert.FileExists(
t, env.archivePath(tgt),
"eviction must not delete the archive file",
)
}
assert.True(
t, env.eng.ExportArchiveHandleOpen(other.ID),
"another webhook's writer must be left alone",
)
}
// TestEvictTarget_LeavesOtherTargets proves that evicting one
// database target leaves the writer of another target of the same
// webhook in place.
func TestEvictTarget_LeavesOtherTargets(t *testing.T) {
t.Parallel()
env := setupArchiveTest(t)
doomed := env.seedDatabaseTarget(t, "")
kept := env.addDatabaseTarget(t, doomed.WebhookID, "")
deliverTo(t, env, doomed)
deliverTo(t, env, kept)
env.eng.EvictTarget(doomed.ID)
assert.False(t, env.eng.ExportHasArchiveWriter(doomed.ID))
assert.FileExists(
t, env.archivePath(doomed),
"eviction must not delete the archive file",
)
assert.True(
t, env.eng.ExportArchiveHandleOpen(kept.ID),
"the other target's writer must be left alone",
)
}
// TestEvictWebhook_UnknownWebhookIsNoOp proves eviction is safe
// for the common case of a webhook or target that never had an
// archive writer, and that repeating it does not panic.
func TestEvictWebhook_UnknownWebhookIsNoOp(t *testing.T) {
t.Parallel()
env := setupArchiveTest(t)
assert.NotPanics(t, func() {
env.eng.EvictWebhook("no-such-webhook")
env.eng.EvictWebhook("no-such-webhook")
env.eng.EvictTarget("no-such-target")
env.eng.EvictTarget("no-such-target")
})
assert.False(
t, env.eng.ExportHasArchiveWriter("no-such-target"),
"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()
env := setupArchiveTest(t)
tgt := env.seedDatabaseTarget(t, "")
// Prime the registry so the test can hold the very writer the
// eviction is about to detach.
deliverTo(t, env, tgt)
w := env.eng.ExportArchiveWriterFor(tgt.ID)
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()
env.eng.EvictWebhook(tgt.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, env.eng.ExportHasArchiveWriter(tgt.ID),
"the registry entry must stay gone",
)
}
// TestEvictWebhook_LaterDeliveryRecreatesWriter proves eviction
// does not break archiving for a target 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()
env := setupArchiveTest(t)
tgt := env.seedDatabaseTarget(t, "")
deliverTo(t, env, tgt)
require.True(t, env.eng.ExportHasArchiveWriter(tgt.ID))
env.eng.EvictWebhook(tgt.WebhookID)
// A fresh delivery for the same target gets a brand new
// writer from the registry, so archiving keeps working.
deliverTo(t, env, tgt)
assert.True(
t, env.eng.ExportHasArchiveWriter(tgt.ID),
"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()
env := setupArchiveTest(t)
tgt := env.seedDatabaseTarget(t, "")
deliverTo(t, env, tgt)
w := env.eng.ExportArchiveWriterFor(tgt.ID)
require.NotNil(t, w)
require.True(t, w.HandleOpen())
require.NoError(t, env.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, env.eng.ExportHasArchiveWriter(tgt.ID),
"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",
)
}
-387
View File
@@ -1,387 +0,0 @@
package delivery
import (
"compress/gzip"
"context"
"database/sql"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"io"
"io/fs"
"log/slog"
"os"
"path/filepath"
"sync"
"time"
"unicode/utf8"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
"sneak.berlin/go/webhooker/internal/database"
"sneak.berlin/go/webhooker/internal/gormlog"
)
// archiveTableQuery counts the archive's table: 0 when the archive
// writer has created the file but not yet the table in it.
const archiveTableQuery = "SELECT count(*) FROM sqlite_master " +
"WHERE type = 'table' AND name = 'archived_events'"
// ArchivePath returns where a database target's archive file is: in
// the data directory, beside the webhook's event database, under the
// name ArchiveFileName gives it.
func ArchivePath(
dbMgr *database.WebhookDBManager,
webhook *database.Webhook,
target *database.Target,
) string {
return filepath.Join(
filepath.Dir(dbMgr.DBPath(webhook.ID)),
ArchiveFileName(webhook.Name, target.Name, target.ID),
)
}
// ArchiveExportFileName returns the name a database target's archive
// downloads under:
// archive-WEBHOOKNAME-TARGETNAME-YYYYMMDDTHHMMSSZ.json.gz, the names
// made safe as in ArchiveFileName and the time in UTC.
func ArchiveExportFileName(
webhookName, targetName string, at time.Time,
) string {
return "archive-" + archiveNamePart(webhookName) + "-" +
archiveNamePart(targetName) + "-" +
at.UTC().Format("20060102T150405Z") + ".json.gz"
}
// ArchiveExport is a database target's archive listed for download. It
// opens one of the target's files at a time, only when its rows are
// about to be written out, and closes it before it opens the next, so
// an export holds at most one file open however many the target has.
//
// Each file is read on its own connection inside one read-only
// transaction, so its rows are written out as the file stood when it
// was opened. Archives are in WAL mode, where a reader works from a
// snapshot and never blocks a writer: archive writes go on while a file
// is open, and the export does not see them. SQLite cannot checkpoint
// a -wal past an open snapshot, so the open file's -wal grows until the
// export has written that file out.
type ArchiveExport struct {
// periods are the periods of the target's files when the export
// was listed, "" for the file without one, in the order
// archiveFiles lists them.
periods []string
// lock is held while currentPath is called and a file is opened,
// so that a rename, which holds it too, cannot move the file in
// between.
lock sync.Locker
// currentPath returns the path ArchivePath gives the target under
// the names stored for it now, which a rename may have changed since
// the export was listed.
currentPath func() (string, error)
log *slog.Logger
}
// exportFile is one archive file opened for an export.
type exportFile struct {
db *sql.DB
tx *gorm.DB
// period is the period in the file's name, "" for none.
period string
// empty is true for a file without the archive's table yet.
empty bool
}
// exportedName is how an export names its webhook and its target.
type exportedName struct {
ID string `json:"id"`
Name string `json:"name"`
}
// NewArchiveExport lists a database target's archive files for export,
// given the path ArchivePath gives it (see archiveFiles). It opens none
// of them. Its caller holds lock, which every rename of the target's
// files runs under, from reading the names path is made of until it
// returns, so the files it lists are the ones those names give.
//
// WriteGzipJSON, called without lock held, finds each file again by its
// period under the path currentPath gives, holding lock while it does
// and while it opens the file, so a rename during the export loses no
// file. A file that is gone by then, emptied by the sweep or moved
// away, is skipped. The export never creates a file: with no files, it
// has no rows.
func NewArchiveExport(
path string,
lock sync.Locker,
currentPath func() (string, error),
log *slog.Logger,
) (*ArchiveExport, error) {
files, err := archiveFiles(path)
if err != nil {
return nil, err
}
x := &ArchiveExport{lock: lock, currentPath: currentPath, log: log}
for _, file := range files {
x.periods = append(x.periods, file.period)
}
return x, nil
}
// openExportFile opens one archive file for an export and takes its
// snapshot. The transaction lasts as long as ctx does.
func openExportFile(
ctx context.Context, file archiveFile, log *slog.Logger,
) (*exportFile, error) {
db, err := database.OpenSQLite(file.path, archiveModeExisting)
if err != nil {
return nil, fmt.Errorf("opening archive %s: %w", file.path, err)
}
gdb, err := gorm.Open(
sqlite.Dialector{Conn: db}, &gorm.Config{
// Never leave this at GORM's default. See
// internal/gormlog.
Logger: gormlog.New(log),
},
)
if err != nil {
_ = db.Close()
return nil, fmt.Errorf("opening archive %s: %w", file.path, err)
}
// ReadOnly makes the driver begin a deferred transaction in place
// of the BEGIN IMMEDIATE the connection string asks for, so the
// export never takes the archive's write lock.
tx := gdb.WithContext(ctx).Begin(&sql.TxOptions{ReadOnly: true})
if tx.Error != nil {
_ = db.Close()
return nil, fmt.Errorf(
"reading archive %s: %w", file.path, tx.Error,
)
}
// The transaction's first read is what takes the snapshot.
var tables int
err = tx.Raw(archiveTableQuery).Row().Scan(&tables)
if err != nil {
_ = tx.Rollback()
_ = db.Close()
return nil, fmt.Errorf("reading archive %s: %w", file.path, err)
}
return &exportFile{
db: db, tx: tx, period: file.period, empty: tables == 0,
}, nil
}
// WriteGzipJSON writes the export to w as one gzipped JSON object:
// webhook and target, each an id and a name; exported_at; and
// archived_events, one object per archived row, keyed by column name,
// the files in the order archiveFiles lists them. A row from a file
// named for a period has "period" beside its columns. A body that is
// not valid UTF-8 cannot be a JSON string, so it is written in base64,
// with "body_encoding": "base64" beside it.
//
// Each row is written out before the next is read, so neither the
// archive nor its JSON is ever held in memory whole, and each file is
// closed once its rows are written, before the next is opened. When it
// returns, no file is open. After an error the gzip stream is left
// unfinished, so what was written does not decompress as a whole file.
func (x *ArchiveExport) WriteGzipJSON(
ctx context.Context,
w io.Writer,
webhook *database.Webhook,
target *database.Target,
exportedAt time.Time,
) error {
head, err := json.Marshal(map[string]any{
"webhook": exportedName{ID: webhook.ID, Name: webhook.Name},
"target": exportedName{ID: target.ID, Name: target.Name},
"exported_at": exportedAt.UTC(),
})
if err != nil {
return fmt.Errorf("encoding archive export: %w", err)
}
zw := gzip.NewWriter(w)
err = x.writeJSON(ctx, zw, head)
if err != nil {
return fmt.Errorf("writing archive export: %w", err)
}
return zw.Close()
}
// openFile finds the target's archive file for period under the path
// currentPath gives now, and opens it for the export, holding x.lock
// for both. For a file that is gone, the error wraps fs.ErrNotExist.
func (x *ArchiveExport) openFile(
ctx context.Context, period string,
) (*exportFile, error) {
x.lock.Lock()
defer x.lock.Unlock()
path, err := x.currentPath()
if err != nil {
return nil, fmt.Errorf("finding archive file: %w", err)
}
file := archiveFile{path: archivePeriodPath(path, period), period: period}
_, err = os.Stat(file.path)
if err != nil {
return nil, err
}
return openExportFile(ctx, file, x.log)
}
// close ends the file's transaction and closes its connection.
func (f *exportFile) close() error {
_ = f.tx.Rollback()
return f.db.Close()
}
// writeJSON writes head with archived_events added as its last key,
// the rows going into it one at a time.
func (x *ArchiveExport) writeJSON(
ctx context.Context, w io.Writer, head []byte,
) error {
// head goes out without its closing brace, so that
// archived_events can follow it.
_, err := w.Write(head[:len(head)-1])
if err != nil {
return err
}
_, err = io.WriteString(w, `,"archived_events":[`)
if err != nil {
return err
}
err = x.writeRows(ctx, w)
if err != nil {
return err
}
_, err = io.WriteString(w, "\n]}\n")
return err
}
// writeRows writes the archived rows of each file to w, one per line,
// separated by commas, opening each file in turn and closing it once
// its rows are written.
func (x *ArchiveExport) writeRows(ctx context.Context, w io.Writer) error {
sep := "\n"
for _, period := range x.periods {
f, err := x.openFile(ctx, period)
if errors.Is(err, fs.ErrNotExist) {
continue
}
if err != nil {
return err
}
sep, err = f.writeRows(ctx, w, sep)
err = errors.Join(err, f.close())
if err != nil {
return err
}
}
return nil
}
// writeRows writes the file's archived rows to w, oldest first, the
// first after sep and each other after ",\n". It returns what goes
// before the next row: sep again when the file had no rows.
func (f *exportFile) writeRows(
ctx context.Context, w io.Writer, sep string,
) (string, error) {
if f.empty {
return sep, nil
}
rows, err := f.tx.WithContext(ctx).
Model(&archivedEvent{}).Order("id").Rows()
if err != nil {
return "", err
}
defer func() { _ = rows.Close() }()
for ; rows.Next(); sep = ",\n" {
var ev archivedEvent
err = f.tx.ScanRows(rows, &ev)
if err != nil {
return "", err
}
_, err = io.WriteString(w, sep)
if err != nil {
return "", err
}
err = writeRow(w, &ev, f.period)
if err != nil {
return "", err
}
}
return sep, rows.Err()
}
// writeRow writes an archived row to w as a JSON object keyed by
// column name, its body in base64 when it is not valid UTF-8, with
// the period of its file beside them unless that is "".
func writeRow(w io.Writer, ev *archivedEvent, period string) error {
row := map[string]any{
"id": ev.ID,
"event_id": ev.EventID,
"webhook_id": ev.WebhookID,
"entrypoint_id": ev.EntrypointID,
"method": ev.Method,
"raw_query": ev.RawQuery,
"headers": ev.Headers,
"body": ev.Body,
"content_type": ev.ContentType,
"archived_at": ev.ArchivedAt.UTC(),
}
if !utf8.ValidString(ev.Body) {
row["body"] = base64.StdEncoding.EncodeToString([]byte(ev.Body))
row["body_encoding"] = "base64"
}
if period != "" {
row["period"] = period
}
line, err := json.Marshal(row)
if err != nil {
return err
}
_, err = w.Write(line)
return err
}
@@ -1,573 +0,0 @@
package delivery_test
import (
"bufio"
"bytes"
"compress/gzip"
"crypto/rand"
"encoding/base64"
"encoding/json"
"fmt"
"io"
"os"
"path/filepath"
"runtime"
"strings"
"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"
)
// The webhook and the target the export tests' archives belong to.
const (
exportWebhookID = "wh-export"
exportWebhookName = "Orders (EU)"
exportTargetID = "tgt-export"
exportTargetName = "Long-term archive"
)
// binaryBody is a body that is not valid UTF-8.
const binaryBody = "\xff\xfe\x00\x01binary\x80"
// writeExportTo writes export to w as the archive of the export tests'
// webhook and target, exported at 2026-10-02T12:03:04Z.
func writeExportTo(
t *testing.T, export *delivery.ArchiveExport, w io.Writer,
) error {
t.Helper()
return export.WriteGzipJSON(
t.Context(), w,
&database.Webhook{
BaseModel: database.BaseModel{ID: exportWebhookID},
Name: exportWebhookName,
},
&database.Target{
BaseModel: database.BaseModel{ID: exportTargetID},
Name: exportTargetName,
},
time.Date(2026, 10, 2, 12, 3, 4, 0, time.UTC),
)
}
// listExport lists the archive at path for export, as the archive of a
// target whose names do not change.
func listExport(t *testing.T, path string) *delivery.ArchiveExport {
t.Helper()
return newExport(t, path, &sync.Mutex{}, func() (string, error) {
return path, nil
})
}
// newExport lists the archive at path for export, to find each file
// again under the path currentPath gives, holding lock while it does.
// Nothing else takes lock while it lists, so it does not hold lock.
func newExport(
t *testing.T,
path string,
lock sync.Locker,
currentPath func() (string, error),
) *delivery.ArchiveExport {
t.Helper()
export, err := delivery.NewArchiveExport(
path, lock, currentPath, archiveTestLogger(),
)
require.NoError(t, err)
return export
}
// exportArchive runs a whole export of the archive at path and returns
// its JSON, decompressed and parsed.
func exportArchive(t *testing.T, path string) map[string]any {
t.Helper()
return writeExport(t, listExport(t, path))
}
// writeExport writes an opened export and returns its JSON,
// decompressed and parsed. Reading to the end makes the gzip reader
// check that the stream was finished.
func writeExport(
t *testing.T, export *delivery.ArchiveExport,
) map[string]any {
t.Helper()
var buf bytes.Buffer
require.NoError(t, writeExportTo(t, export, &buf))
zr, err := gzip.NewReader(&buf)
require.NoError(t, err)
raw, err := io.ReadAll(zr)
require.NoError(t, err)
var got map[string]any
require.NoError(t, json.Unmarshal(raw, &got))
return got
}
// exportedEvents returns an export's archived_events.
func exportedEvents(t *testing.T, got map[string]any) []map[string]any {
t.Helper()
list, ok := got["archived_events"].([]any)
require.True(t, ok, "archived_events must be an array: %v", got)
events := make([]map[string]any, len(list))
for i, v := range list {
events[i], ok = v.(map[string]any)
require.True(t, ok, "an archived event must be an object: %v", v)
}
return events
}
// exportedEventIDs returns the event_id of each of an export's
// archived_events.
func exportedEventIDs(t *testing.T, got map[string]any) []string {
t.Helper()
events := exportedEvents(t, got)
ids := make([]string, 0, len(events))
for _, ev := range events {
ids = append(ids, fmt.Sprint(ev["event_id"]))
}
return ids
}
// TestArchiveExport_MatchesStoredRows proves an export holds the
// webhook, the target, the time, and every column of every stored
// row: a body that is valid UTF-8 as a string, and one that is not in
// base64, marked as such.
func TestArchiveExport_MatchesStoredRows(t *testing.T) {
t.Parallel()
path := filepath.Join(t.TempDir(), "archive.db")
w := delivery.NewExportArchiveWriter(path, archiveTestLogger(), 0)
bodies := []string{`{"order":1}`, "plain text", "", binaryBody}
for i, body := range bodies {
require.NoError(t, w.Write(delivery.ExportArchivedEvent{
EventID: fmt.Sprintf("ev-%d", i),
WebhookID: exportWebhookID,
EntrypointID: "ep-1",
Method: "POST",
RawQuery: eventQuery,
Headers: `{"X-Test":["yes"]}`,
Body: body,
ContentType: testContentType,
}, 0))
}
var stored []delivery.ExportArchivedEvent
require.NoError(t, openArchiveDBForRead(t, path).
Order("id").Find(&stored).Error)
got := exportArchive(t, path)
assert.Equal(t,
map[string]any{"id": exportWebhookID, "name": exportWebhookName},
got["webhook"],
)
assert.Equal(t,
map[string]any{"id": exportTargetID, "name": exportTargetName},
got["target"],
)
assert.Equal(t, "2026-10-02T12:03:04Z", got["exported_at"])
events := exportedEvents(t, got)
require.Len(t, events, len(bodies))
for i, row := range stored {
assertExportedRow(t, row, events[i])
}
}
// assertExportedRow checks that ev, from an export, holds every column
// of the stored row.
func assertExportedRow(
t *testing.T, row delivery.ExportArchivedEvent, ev map[string]any,
) {
t.Helper()
archivedAt, err := time.Parse(
time.RFC3339Nano, fmt.Sprint(ev["archived_at"]),
)
require.NoError(t, err)
assert.True(t, archivedAt.Equal(row.ArchivedAt))
assert.EqualValues(t, row.ID, ev["id"])
assert.Equal(t, row.EventID, ev["event_id"])
assert.Equal(t, row.WebhookID, ev["webhook_id"])
assert.Equal(t, row.EntrypointID, ev["entrypoint_id"])
assert.Equal(t, row.Method, ev["method"])
assert.Equal(t, row.RawQuery, ev["raw_query"])
assert.Equal(t, row.Headers, ev["headers"])
assert.Equal(t, row.ContentType, ev["content_type"])
if row.Body != binaryBody {
assert.Equal(t, row.Body, ev["body"])
assert.Len(t, ev, 10, "the ten columns and nothing else: %v", ev)
return
}
body, err := base64.StdEncoding.DecodeString(fmt.Sprint(ev["body"]))
require.NoError(t, err)
assert.Equal(t, binaryBody, string(body))
assert.Equal(t, "base64", ev["body_encoding"])
assert.Len(t, ev, 11, "the ten columns and body_encoding: %v", ev)
}
// TestArchiveExport_Empty proves an archive with nothing in it exports
// as an empty archived_events: no file, which the export must not
// create; a file the archive writer has not yet put its table in; and
// a table with no rows.
func TestArchiveExport_Empty(t *testing.T) {
t.Parallel()
dir := t.TempDir()
missing := filepath.Join(dir, "missing.db")
noTable := filepath.Join(dir, "no-table.db")
noRows := filepath.Join(dir, "no-rows.db")
require.NoError(t, os.WriteFile(noTable, nil, 0o600))
require.NoError(t,
delivery.NewExportArchiveWriter(noRows, archiveTestLogger(), 0).
Open(0),
)
for _, path := range []string{missing, noTable, noRows} {
assert.Empty(t, exportedEvents(t, exportArchive(t, path)), path)
}
for _, suffix := range archiveFileSuffixes() {
assert.NoFileExists(t, missing+suffix)
}
}
// unlockHook is a sync.Locker that runs fn each time it is unlocked. An
// export unlocks its lock right after it opens a file.
type unlockHook struct {
sync.Mutex
fn func()
}
func (u *unlockHook) Unlock() {
u.Mutex.Unlock()
u.fn()
}
// TestArchiveExport_ReadsOneSnapshot proves an export writes a file
// out as it was when the export opened it, and holds up no archive
// write: a row written after the export was listed but before the file
// was opened is in the export, and one written while the file is open
// is stored, and is not. A write held up for the whole busy timeout
// would fail.
func TestArchiveExport_ReadsOneSnapshot(t *testing.T) {
t.Parallel()
path := filepath.Join(t.TempDir(), "archive.db")
w := delivery.NewExportArchiveWriter(path, archiveTestLogger(), 0)
require.NoError(t, w.Write(delivery.ExportArchivedEvent{EventID: "listed"}, 0))
opened := &unlockHook{fn: func() {
require.NoError(t, w.Write(delivery.ExportArchivedEvent{EventID: "during"}, 0))
}}
export := newExport(t, path, opened, func() (string, error) {
return path, nil
})
require.NoError(t, w.Write(delivery.ExportArchivedEvent{EventID: "before-open"}, 0))
assert.Equal(t,
[]string{"listed", "before-open"},
exportedEventIDs(t, writeExport(t, export)),
)
var stored int64
require.NoError(t, openArchiveDBForRead(t, path).
Model(&delivery.ExportArchivedEvent{}).Count(&stored).Error)
assert.Equal(t, int64(3), stored)
}
// TestArchiveExport_FindsFilesAfterRename proves that renaming the
// archive after an export has listed it, as renaming its webhook or
// target does, loses no file: the export finds each file again by its
// period under the new name. A file moved away by then is skipped.
func TestArchiveExport_FindsFilesAfterRename(t *testing.T) {
t.Parallel()
dir := t.TempDir()
path := filepath.Join(dir, "archive-old.db")
w := delivery.NewExportArchiveWriter(path, archiveTestLogger(), 0)
for _, period := range []string{"", dayPeriod, hourPeriod} {
require.NoError(t, w.WritePeriod(
delivery.ExportArchivedEvent{EventID: "in-" + period}, 0, period,
))
}
current := path
export := newExport(t, path, &sync.Mutex{}, func() (string, error) {
return current, nil
})
require.NoError(t, w.Rename("archive-new.db"))
current = filepath.Join(dir, "archive-new.db")
removeArchiveFiles(t, periodPath(current, dayPeriod))
assert.Equal(t,
[]string{"in-", "in-" + hourPeriod},
exportedEventIDs(t, writeExport(t, export)),
)
}
// TestArchiveExport_EveryFileOldestFirst writes a row to a target's
// file without a period and to its files for a month, an hour and a
// day, and proves the export holds every row: the file without a
// period first, then the others oldest period first, each row from a
// file named for a period carrying that period. A file made after the
// export was listed is not in it.
func TestArchiveExport_EveryFileOldestFirst(t *testing.T) {
t.Parallel()
path := filepath.Join(t.TempDir(), "archive-wh.db")
w := delivery.NewExportArchiveWriter(path, archiveTestLogger(), 0)
// Written in an order that is not the export's.
for _, period := range []string{nextDayPeriod, "", hourPeriod, "2026-03"} {
require.NoError(t, w.WritePeriod(
delivery.ExportArchivedEvent{EventID: "in-" + period}, 0, period,
))
}
export := listExport(t, path)
require.NoError(t, w.WritePeriod(
delivery.ExportArchivedEvent{EventID: "later"}, 0, "2026-03-06",
))
events := exportedEvents(t, writeExport(t, export))
ids := make([]string, 0, len(events))
periods := make([]any, 0, len(events))
for _, ev := range events {
ids = append(ids, fmt.Sprint(ev["event_id"]))
periods = append(periods, ev["period"])
}
assert.Equal(t,
[]string{"in-", "in-2026-03", "in-" + hourPeriod, "in-" + nextDayPeriod},
ids,
)
assert.Equal(t,
[]any{nil, "2026-03", hourPeriod, nextDayPeriod}, periods,
)
assert.NotContains(t, events[0], "period",
"a row from the file without a period has no period")
}
// openFilesPeak is an io.Writer that discards what it is given and
// records the most archive files in dir the process had open at any
// write, as /proc/self/fd lists the files a process has open.
type openFilesPeak struct {
dir string
max int
}
func (p *openFilesPeak) Write(b []byte) (int, error) {
fds, err := os.ReadDir("/proc/self/fd")
if err != nil {
return 0, err
}
open := map[string]bool{}
for _, fd := range fds {
file, err := os.Readlink(filepath.Join("/proc/self/fd", fd.Name()))
if err == nil && filepath.Dir(file) == p.dir &&
strings.HasSuffix(file, ".db") {
open[file] = true
}
}
p.max = max(p.max, len(open))
return len(b), nil
}
// TestArchiveExport_OneFileOpenAtATime exports a target with a file for
// each of 24 hours and proves the export never had more than one of
// them open, and had one open while it wrote. Each file holds a row of
// 48 KiB of random base64, which gzip shrinks little, so the export
// writes output while it reads each file.
func TestArchiveExport_OneFileOpenAtATime(t *testing.T) {
t.Parallel()
if runtime.GOOS != "linux" {
t.Skip("only Linux lists a process's open files in /proc/self/fd")
}
// Readlink gives each open file's path with no symbolic link in it.
dir, err := filepath.EvalSymlinks(t.TempDir())
require.NoError(t, err)
path := filepath.Join(dir, "archive-wh.db")
w := delivery.NewExportArchiveWriter(path, archiveTestLogger(), 0)
random := make([]byte, 36<<10)
for hour := range 24 {
_, _ = rand.Read(random)
require.NoError(t, w.WritePeriod(delivery.ExportArchivedEvent{
Body: base64.StdEncoding.EncodeToString(random),
}, 0, fmt.Sprintf("2026-10-01-%02d", hour)))
}
// The writer's own handle on the last file is not the export's.
w.Evict()
// Through a buffer, the open files are listed once per 8 KiB of
// output, a few times for each file, rather than at each of gzip's
// small writes, which takes far longer.
peak := &openFilesPeak{dir: dir}
buffered := bufio.NewWriterSize(peak, 8<<10)
require.NoError(t, writeExportTo(t, listExport(t, path), buffered))
require.NoError(t, buffered.Flush())
assert.Equal(t, 1, peak.max)
}
// heapPeak is an io.Writer that discards what it is given and records
// the largest heap it saw at a write. It collects garbage twice before
// each reading, so the heap it reads is what is still held. Once is not
// enough: the libraries the export calls (regexp, under GORM's table
// names, and encoding/json among them) cache buffers in a sync.Pool,
// which keeps them through one collection, so after one the reading
// counts however many happen to be cached. That varies from run to run
// by about as much as the limit in TestArchiveExport_Streams.
type heapPeak struct {
max uint64
}
func (p *heapPeak) Write(b []byte) (int, error) {
var m runtime.MemStats
runtime.GC()
runtime.GC()
runtime.ReadMemStats(&m)
p.max = max(p.max, m.HeapAlloc)
return len(b), nil
}
// exportHeapGrowth exports an archive of rows random bodies, each
// bodySize bytes of base64, and returns how far the heap rose above
// where it stood when the export began, at its highest.
func exportHeapGrowth(t *testing.T, rows, bodySize int) uint64 {
t.Helper()
path := filepath.Join(t.TempDir(), "archive.db")
w := delivery.NewExportArchiveWriter(path, archiveTestLogger(), 0)
// Base64 makes four characters of every three bytes.
random := make([]byte, bodySize/4*3)
for range rows {
_, _ = rand.Read(random)
require.NoError(t, w.Write(delivery.ExportArchivedEvent{
Body: base64.StdEncoding.EncodeToString(random),
}, 0))
}
export := listExport(t, path)
// Twice, for the reason heapPeak gives.
runtime.GC()
runtime.GC()
var start runtime.MemStats
runtime.ReadMemStats(&start)
// Through a buffer, the heap is read once per 8 KiB of output
// rather than at each of gzip's small writes, which takes far
// longer.
peak := &heapPeak{max: start.HeapAlloc}
buffered := bufio.NewWriterSize(peak, 8<<10)
require.NoError(t, writeExportTo(t, export, buffered))
require.NoError(t, buffered.Flush())
return peak.max - start.HeapAlloc
}
// TestArchiveExport_Streams proves an export holds neither the archive
// nor its output in memory whole: exporting 384 KiB more of archive
// raises the heap's peak by less than half of that. The export's own
// memory, mostly gzip's compressor, is the same for both archives, so
// it cancels out. The bodies are random bytes in base64, which gzip
// shrinks by only a quarter, so an export that read every row before
// writing, or built the JSON or the gzipped file before writing it,
// would raise the peak by at least three quarters of the difference.
//
// The smaller archive has two rows so that its export, too, writes
// out more than the 8 KiB buffer in exportHeapGrowth before it ends:
// the heap must be read while the export's own memory is held.
//
//nolint:paralleltest // It measures the heap, which tests share.
func TestArchiveExport_Streams(t *testing.T) {
const (
bodySize = 16 << 10
smallRows = 2
largeRows = smallRows + 24
limit = (largeRows - smallRows) * bodySize / 2
)
small := exportHeapGrowth(t, smallRows, bodySize)
large := exportHeapGrowth(t, largeRows, bodySize)
assert.Less(t, large, small+limit,
"the heap rose by %d for %d rows and by %d for %d rows",
small, smallRows, large, largeRows,
)
}
// TestArchiveExportFileName proves the download is named for the
// webhook and the target, with the names made safe as for the archive
// file, and the export time in UTC.
func TestArchiveExportFileName(t *testing.T) {
t.Parallel()
cest := time.FixedZone("CEST", int((2 * time.Hour).Seconds()))
assert.Equal(t,
"archive-orders-eu-long-term-archive-20261002T120304Z.json.gz",
delivery.ArchiveExportFileName(
exportWebhookName, exportTargetName,
time.Date(2026, 10, 2, 14, 3, 4, 0, cest),
),
)
}
@@ -1,198 +0,0 @@
package delivery
import (
"encoding/json"
"errors"
"fmt"
"os"
"path/filepath"
"slices"
"strings"
"time"
"sneak.berlin/go/webhooker/internal/database"
)
// The archive rotations: how often a database target starts a new
// archive file. Every rotation but none puts the period of an event's
// receive time, in UTC, in the name of the file the event goes to,
// written in the layout beside it.
const (
archiveRotationNone = "none"
archiveRotationMonthly = "monthly"
archiveRotationDaily = "daily"
archiveRotationHourly = "hourly"
archiveMonthLayout = "2006-01"
archiveDayLayout = "2006-01-02"
archiveHourLayout = "2006-01-02-15"
)
// errArchiveRotationUnknown is returned for a rotation that is not one
// of the four.
var errArchiveRotationUnknown = errors.New(
"rotation must be none, monthly, daily or hourly",
)
// archiveFile is one of a database target's archive files, and the
// period in its name: "" for the file named without one.
type archiveFile struct {
path string
period string
}
// ValidateArchiveRotation checks a user-supplied archive rotation for
// a database target: empty or none (both meaning one file), monthly,
// daily or hourly.
func ValidateArchiveRotation(rotation string) error {
switch rotation {
case "", archiveRotationNone, archiveRotationMonthly,
archiveRotationDaily, archiveRotationHourly:
return nil
default:
return fmt.Errorf("%w: %q", errArchiveRotationUnknown, rotation)
}
}
// parseArchiveRotation reads the rotation from a database target's
// config JSON. An empty config or an empty rotation is none.
func parseArchiveRotation(configJSON string) (string, error) {
if configJSON == "" {
return archiveRotationNone, nil
}
var cfg databaseTargetConfig
err := json.Unmarshal([]byte(configJSON), &cfg)
if err != nil {
return "", fmt.Errorf("parsing database target config: %w", err)
}
err = ValidateArchiveRotation(cfg.Rotation)
if err != nil {
return "", err
}
if cfg.Rotation == "" {
return archiveRotationNone, nil
}
return cfg.Rotation, nil
}
// archivePeriod returns the period, in UTC, that a rotation puts an
// event received at receivedAt in: "2026-10" for monthly,
// "2026-10-01" for daily, "2026-10-01-19" for hourly, and "" for none.
func archivePeriod(rotation string, receivedAt time.Time) string {
switch rotation {
case archiveRotationMonthly:
return receivedAt.UTC().Format(archiveMonthLayout)
case archiveRotationDaily:
return receivedAt.UTC().Format(archiveDayLayout)
case archiveRotationHourly:
return receivedAt.UTC().Format(archiveHourLayout)
default:
return ""
}
}
// archivePeriodPath returns the path of a database target's archive
// file for a period: path, as ArchivePath gives it, with "-" and the
// period put before its ".db". The period "" gives path itself.
func archivePeriodPath(path, period string) string {
if period == "" {
return path
}
return strings.TrimSuffix(path, ".db") + "-" + period + ".db"
}
// ArchivePathAt returns the archive file a database target writes an
// event received at receivedAt to: ArchivePath's file, with the
// period in its name when the target rotates.
func ArchivePathAt(
dbMgr *database.WebhookDBManager,
webhook *database.Webhook,
target *database.Target,
receivedAt time.Time,
) (string, error) {
rotation, err := parseArchiveRotation(target.Config)
if err != nil {
return "", err
}
return archivePeriodPath(
ArchivePath(dbMgr, webhook, target),
archivePeriod(rotation, receivedAt),
), nil
}
// archiveFiles lists the database target's archive files that exist,
// given the path ArchivePath gives it: the file at path, then each
// file archivePeriodPath names for path and a period, oldest period
// first. Which rotation wrote a file does not matter, so the files of
// an earlier rotation setting are listed too.
func archiveFiles(path string) ([]archiveFile, error) {
dir := filepath.Dir(path)
entries, err := os.ReadDir(dir)
if err != nil {
return nil, fmt.Errorf("listing archive files: %w", err)
}
stem := strings.TrimSuffix(filepath.Base(path), ".db")
var files []archiveFile
for _, entry := range entries {
period, ok := archiveFilePeriod(stem, entry.Name())
if ok {
files = append(files, archiveFile{
path: filepath.Join(dir, entry.Name()),
period: period,
})
}
}
// A month sorts before the days and hours in it.
slices.SortFunc(files, func(a, b archiveFile) int {
return strings.Compare(a.period, b.period)
})
return files, nil
}
// archiveFilePeriod reports whether name is the name of an archive
// file of the target whose file name without a period is stem+".db",
// and the period in it.
func archiveFilePeriod(stem, name string) (string, bool) {
rest, ok := strings.CutPrefix(name, stem)
if !ok {
return "", false
}
rest, ok = strings.CutSuffix(rest, ".db")
if !ok {
return "", false
}
if rest == "" {
return "", true
}
period, ok := strings.CutPrefix(rest, "-")
if !ok {
return "", false
}
for _, layout := range []string{
archiveMonthLayout, archiveDayLayout, archiveHourLayout,
} {
_, err := time.Parse(layout, period)
if err == nil {
return period, true
}
}
return "", false
}
@@ -1,390 +0,0 @@
package delivery_test
import (
"os"
"path/filepath"
"strings"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"sneak.berlin/go/webhooker/internal/database"
"sneak.berlin/go/webhooker/internal/database/databasetest"
"sneak.berlin/go/webhooker/internal/delivery"
)
// The archive rotations, and the configs of a daily target and of one
// whose rotation is not one of the four.
const (
rotationNone = "none"
rotationMonthly = "monthly"
rotationDaily = "daily"
rotationHourly = "hourly"
dailyConfig = `{"rotation":"daily"}`
weeklyConfig = `{"rotation":"weekly"}`
)
// The periods the tests archive into most: two days, and an hour of
// the first.
const (
dayPeriod = "2026-03-04"
nextDayPeriod = "2026-03-05"
hourPeriod = "2026-03-04-05"
)
// periodPath returns the archive file for a period of the target whose
// file without a period is path.
func periodPath(path, period string) string {
return strings.TrimSuffix(path, ".db") + "-" + period + ".db"
}
// deliverReceivedAt delivers to a database target an event whose
// receive time is receivedAt, and returns the event's id. The receive
// time is what decides a rotated archive's file, so setting it is how
// these tests move the clock across a period boundary.
func (env *archiveEnv) deliverReceivedAt(
t *testing.T, tgt *database.Target, receivedAt time.Time,
) string {
t.Helper()
webhookDB := testWebhookDB(t)
event := seedEvent(t, webhookDB, `{"n":1}`)
event.CreatedAt = receivedAt
env.eng.ExportDeliverDatabase(
webhookDB, seedDatabaseTargetDelivery(t, webhookDB, event, tgt),
)
return event.ID
}
// TestDeliverDatabase_RotatesAtEachPeriodBoundary delivers, for each
// rotation, an event received in the last second of a period and one
// received in the first second of the next, and checks each lands in
// the file named for its own period, in UTC. Rotation none keeps both
// in the one file.
func TestDeliverDatabase_RotatesAtEachPeriodBoundary(t *testing.T) {
t.Parallel()
berlin := time.FixedZone("CEST", 2*60*60)
cases := []struct {
name string
rotation string
before, after time.Time
// periods are the periods of before and after.
periods [2]string
}{
{
rotationMonthly, rotationMonthly,
time.Date(2026, 1, 31, 23, 59, 59, 0, time.UTC),
time.Date(2026, 2, 1, 0, 0, 0, 0, time.UTC),
[2]string{"2026-01", "2026-02"},
},
{
rotationDaily, rotationDaily,
time.Date(2026, 3, 4, 23, 59, 59, 0, time.UTC),
time.Date(2026, 3, 5, 0, 0, 0, 0, time.UTC),
[2]string{dayPeriod, nextDayPeriod},
},
{
// The same instants, received in a zone two hours ahead
// of UTC, where they fall on 5 March: the period is UTC's.
"daily in another zone", rotationDaily,
time.Date(2026, 3, 5, 1, 59, 59, 0, berlin),
time.Date(2026, 3, 5, 2, 0, 0, 0, berlin),
[2]string{dayPeriod, nextDayPeriod},
},
{
rotationHourly, rotationHourly,
time.Date(2026, 3, 4, 5, 59, 59, 0, time.UTC),
time.Date(2026, 3, 4, 6, 0, 0, 0, time.UTC),
[2]string{hourPeriod, "2026-03-04-06"},
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
env := setupArchiveTest(t)
tgt := env.seedDatabaseTarget(t, `{"rotation":"`+tc.rotation+`"}`)
path := env.archivePath(tgt)
first := env.deliverReceivedAt(t, tgt, tc.before)
second := env.deliverReceivedAt(t, tgt, tc.after)
assert.Equal(t, []string{first},
archivedEventIDs(t, periodPath(path, tc.periods[0])))
assert.Equal(t, []string{second},
archivedEventIDs(t, periodPath(path, tc.periods[1])))
assert.NoFileExists(t, path,
"a rotated target never writes the file without a period")
})
}
t.Run(rotationNone, func(t *testing.T) {
t.Parallel()
env := setupArchiveTest(t)
tgt := env.seedDatabaseTarget(t, `{"rotation":"`+rotationNone+`"}`)
first := env.deliverReceivedAt(t, tgt, cases[0].before)
second := env.deliverReceivedAt(t, tgt, cases[0].after)
assert.ElementsMatch(t, []string{first, second},
archivedEventIDs(t, env.archivePath(tgt)))
})
}
// TestDeliverDatabase_EventWithoutReceiveTime proves an event whose
// receive time is not known goes to the file for the time it is
// archived, rather than to one for the year 1.
func TestDeliverDatabase_EventWithoutReceiveTime(t *testing.T) {
t.Parallel()
env := setupArchiveTest(t)
tgt := env.seedDatabaseTarget(t, `{"rotation":"`+rotationMonthly+`"}`)
path := env.archivePath(tgt)
before := time.Now().UTC().Format("2006-01")
id := env.deliverReceivedAt(t, tgt, time.Time{})
file := periodPath(path, time.Now().UTC().Format("2006-01"))
_, err := os.Stat(file)
if err != nil {
// The month turned during the delivery.
file = periodPath(path, before)
}
assert.Equal(t, []string{id}, archivedEventIDs(t, file))
assert.NoFileExists(t, periodPath(path, "0001-01"))
}
// TestDeliverDatabase_RotationChangeKeepsOldFiles changes a target's
// rotation from none to daily between two events, and checks the
// second goes to the daily file while the first stays where it was.
func TestDeliverDatabase_RotationChangeKeepsOldFiles(t *testing.T) {
t.Parallel()
env := setupArchiveTest(t)
tgt := env.seedDatabaseTarget(t, "")
path := env.archivePath(tgt)
at := time.Date(2026, 3, 4, 12, 0, 0, 0, time.UTC)
first := env.deliverReceivedAt(t, tgt, at)
tgt.Config = dailyConfig
second := env.deliverReceivedAt(t, tgt, at)
assert.Equal(t, []string{first}, archivedEventIDs(t, path))
assert.Equal(t, []string{second},
archivedEventIDs(t, periodPath(path, dayPeriod)))
}
// TestArchiveSweep_PrunesEveryFile gives a daily target three files:
// the file without a period, left from before it rotated, and two
// daily files. Each holds a row older than the expiry, and one daily
// file also a newer row. The sweep prunes the old row from every file,
// deletes the daily file it leaves empty, and keeps the file without a
// period although it is empty too.
func TestArchiveSweep_PrunesEveryFile(t *testing.T) {
t.Parallel()
env := setupArchiveTest(t)
tgt := env.seedDatabaseTarget(
t, `{"expiry":"1h","rotation":"`+rotationDaily+`"}`,
)
path := env.archivePath(tgt)
emptied := periodPath(path, dayPeriod)
kept := periodPath(path, nextDayPeriod)
now := time.Now()
old := now.Add(-48 * time.Hour)
seedArchiveFile(t, path, tgt.WebhookID, old)
seedArchiveFile(t, emptied, tgt.WebhookID, old)
seedArchiveFile(t, kept, tgt.WebhookID, old, now.Add(-time.Minute))
env.sweeper.ExportSweep(t.Context())
assert.Empty(t, archivedEventIDs(t, path))
assert.Equal(t, []string{sweepRowNew}, archivedEventIDs(t, kept))
for _, suffix := range archiveFileSuffixes() {
assert.NoFileExists(t, emptied+suffix)
}
}
// TestRename_MovesEveryFile renames a daily target that also has a
// file without a period, and checks every file moves to the new name
// with its period, rows and all, and that a later write uses the new
// name.
func TestRename_MovesEveryFile(t *testing.T) {
t.Parallel()
env := setupArchiveTest(t)
tgt := env.seedDatabaseTarget(t, "")
oldPath := env.archivePath(tgt)
day := time.Date(2026, 3, 4, 12, 0, 0, 0, time.UTC)
unrotated := env.deliverReceivedAt(t, tgt, day)
tgt.Config = dailyConfig
first := env.deliverReceivedAt(t, tgt, day)
second := env.deliverReceivedAt(t, tgt, day.Add(24*time.Hour))
require.NoError(t, env.eng.Rename(tgt.ID, "Orders", "Long Term"))
newPath := filepath.Join(
env.dataDir, "archive-orders-long-term-"+tgt.ID+".db",
)
for _, old := range []string{
oldPath,
periodPath(oldPath, dayPeriod),
periodPath(oldPath, nextDayPeriod),
} {
assert.NoFileExists(t, old)
}
assert.Equal(t, []string{unrotated}, archivedEventIDs(t, newPath))
assert.Equal(t, []string{first},
archivedEventIDs(t, periodPath(newPath, dayPeriod)))
assert.Equal(t, []string{second},
archivedEventIDs(t, periodPath(newPath, nextDayPeriod)))
third := env.deliverReceivedAt(t, tgt, day.Add(48*time.Hour))
assert.Equal(t, []string{third},
archivedEventIDs(t, periodPath(newPath, "2026-03-06")))
}
// TestRename_NeverReplacesARotatedFile plants a file at the new name
// of a target's daily file, and proves the rename is refused and moves
// none of the target's files.
func TestRename_NeverReplacesARotatedFile(t *testing.T) {
t.Parallel()
env := setupArchiveTest(t)
tgt := env.seedDatabaseTarget(t, dailyConfig)
oldPath := env.archivePath(tgt)
day := time.Date(2026, 3, 4, 12, 0, 0, 0, time.UTC)
env.deliverReceivedAt(t, tgt, day)
env.deliverReceivedAt(t, tgt, day.Add(24*time.Hour))
newPath := filepath.Join(
env.dataDir, "archive-orders-long-term-"+tgt.ID+".db",
)
planted := periodPath(newPath, nextDayPeriod)
require.NoError(t, os.WriteFile(planted, []byte("planted"), 0o600))
require.ErrorIs(
t, env.eng.Rename(tgt.ID, "Orders", "Long Term"),
delivery.ErrArchiveNameTaken,
)
assert.FileExists(t, periodPath(oldPath, dayPeriod))
assert.FileExists(t, periodPath(oldPath, nextDayPeriod))
assert.NoFileExists(t, periodPath(newPath, dayPeriod))
}
// TestStatArchive_EveryFile proves StatArchive counts and adds up every
// one of a target's files, takes the latest write of any of them, and
// leaves out files whose names only look like the target's.
func TestStatArchive_EveryFile(t *testing.T) {
t.Parallel()
dir := t.TempDir()
path := filepath.Join(dir, "archive-wh.db")
files := []string{
path, periodPath(path, "2026-03"), periodPath(path, hourPeriod),
}
written := time.Date(2026, 3, 4, 5, 6, 7, 0, time.UTC)
var size int64
for i, file := range files {
require.NoError(t, os.WriteFile(file, make([]byte, 100*(i+1)), 0o600))
size += int64(100 * (i + 1))
at := written.Add(-time.Duration(i) * time.Hour)
require.NoError(t, os.Chtimes(file, at, at))
}
for _, other := range []string{
"archive-wh-2026-13.db", "archive-wh-2026-3.db",
"archive-wh-other.db", "archive-wh-2026-03.json",
"archive-whx.db",
} {
require.NoError(t,
os.WriteFile(filepath.Join(dir, other), []byte("x"), 0o600))
}
got, err := delivery.StatArchive(path)
require.NoError(t, err)
assert.Equal(t, len(files), got.Files)
assert.Equal(t, size, got.Size)
assert.True(t, written.Equal(got.Written), got.Written)
}
// TestArchivePathAt names the file each rotation writes an event to.
func TestArchivePathAt(t *testing.T) {
t.Parallel()
dataDir := t.TempDir()
dbMgr := databasetest.NewWebhookDBManager(t, dataDir)
webhook := &database.Webhook{
BaseModel: database.BaseModel{ID: "wh-id"}, Name: "Orders",
}
at := time.Date(2026, 10, 1, 19, 30, 0, 0, time.UTC)
cases := map[string]string{
"": "",
`{"rotation":"` + rotationNone + `"}`: "",
`{"rotation":"` + rotationMonthly + `"}`: "-2026-10",
dailyConfig: "-2026-10-01",
`{"rotation":"` + rotationHourly + `"}`: "-2026-10-01-19",
}
for config, period := range cases {
target := &database.Target{
BaseModel: database.BaseModel{ID: "tgt-id"},
Name: "Archive",
Config: config,
}
got, err := delivery.ArchivePathAt(dbMgr, webhook, target, at)
require.NoError(t, err, config)
assert.Equal(t,
filepath.Join(
dataDir, "archive-orders-archive-tgt-id"+period+".db",
),
got, config,
)
}
_, err := delivery.ArchivePathAt(dbMgr, webhook, &database.Target{
Config: weeklyConfig,
}, at)
require.Error(t, err)
}
// TestValidateArchiveRotation accepts the four rotations, and empty,
// and refuses anything else.
func TestValidateArchiveRotation(t *testing.T) {
t.Parallel()
for _, ok := range []string{
"", rotationNone, rotationMonthly, rotationDaily, rotationHourly,
} {
require.NoError(t, delivery.ValidateArchiveRotation(ok), ok)
}
for _, bad := range []string{"weekly", "Daily", "hourly "} {
require.Error(t, delivery.ValidateArchiveRotation(bad), bad)
}
}
+160 -624
View File
@@ -3,24 +3,65 @@ package delivery_test
import (
"database/sql"
"fmt"
"io/fs"
"log/slog"
"net/http"
"os"
"path/filepath"
"strings"
"testing"
"time"
"github.com/google/uuid"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
"gorm.io/gorm/clause"
_ "modernc.org/sqlite" // Pure Go SQLite driver.
"sneak.berlin/go/webhooker/internal/archive"
"sneak.berlin/go/webhooker/internal/database"
"sneak.berlin/go/webhooker/internal/database/databasetest"
"sneak.berlin/go/webhooker/internal/delivery"
"sneak.berlin/go/webhooker/internal/gormlog"
)
// archiveTestWebhookName is the name of every webhook
// seedDatabaseTarget creates. It is not safe in a file name as it
// stands, so every archive file name made from it is made safe first.
const archiveTestWebhookName = "Sweep Test!"
// archiveEnv bundles the pieces a database target test drives: a main
// configuration database holding webhooks and targets, a delivery
// engine whose archive writers write into the data directory, and that
// directory.
type archiveEnv struct {
eng *delivery.Engine
mainDB *database.Database
dataDir string
}
func setupArchiveTest(t *testing.T) *archiveEnv {
t.Helper()
dataDir := t.TempDir()
mainDB, err := database.Open(dataDir, slog.New(slog.DiscardHandler))
require.NoError(t, err)
t.Cleanup(func() { _ = mainDB.Close() })
return &archiveEnv{
eng: delivery.NewTestEngineWithDB(
mainDB,
databasetest.NewWebhookDBManager(t, dataDir),
archiveTestLogger(),
&http.Client{Timeout: 5 * time.Second},
1,
),
mainDB: mainDB,
dataDir: dataDir,
}
}
func archiveTestLogger() *slog.Logger {
return slog.New(slog.NewTextHandler(
os.Stderr,
@@ -28,8 +69,85 @@ func archiveTestLogger() *slog.Logger {
))
}
// archivePath returns where the engine keeps a database target's
// archive file, for the names seedDatabaseTarget gave it.
func (env *archiveEnv) archivePath(tgt *database.Target) string {
return filepath.Join(
env.dataDir,
archive.FileName(archiveTestWebhookName, tgt.Name, tgt.ID),
)
}
// seedDatabaseTarget creates a webhook with one database target
// carrying the given target config JSON, and returns the target.
func (env *archiveEnv) seedDatabaseTarget(
t *testing.T, configJSON string,
) *database.Target {
t.Helper()
wh := &database.Webhook{
UserID: uuid.New().String(),
Name: archiveTestWebhookName,
}
require.NoError(
t,
env.mainDB.DB().
Omit(clause.Associations).
Create(wh).Error,
)
tgt := &database.Target{
WebhookID: wh.ID,
Name: "Archive",
Type: database.TargetTypeDatabase,
Active: true,
Config: configJSON,
MaxAttempts: 1,
}
require.NoError(
t,
env.mainDB.DB().
Omit(clause.Associations).
Create(tgt).Error,
)
return tgt
}
// seedDatabaseTargetDelivery seeds a pending delivery of an event
// to a database target and returns the in-memory delivery the
// target handler is invoked with.
func seedDatabaseTargetDelivery(
t *testing.T,
webhookDB *gorm.DB,
event database.Event,
tgt *database.Target,
) *database.Delivery {
t.Helper()
dlv := seedDelivery(
t, webhookDB, event.ID, tgt.ID,
database.DeliveryStatusPending,
)
d := &database.Delivery{
EventID: event.ID,
TargetID: tgt.ID,
Status: database.DeliveryStatusPending,
Event: event,
Target: database.Target{
Name: tgt.Name,
Type: database.TargetTypeDatabase,
Config: tgt.Config,
},
}
d.ID = dlv.ID
return d
}
// openArchiveDBForRead opens an archive file read-only so a
// test can inspect the rows the writer persisted.
// test can inspect the rows the database target archived.
func openArchiveDBForRead(
t *testing.T, path string,
) *gorm.DB {
@@ -52,26 +170,45 @@ 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"}
// 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{Logger: gormlog.New(slog.New(slog.DiscardHandler))},
)
if err != nil {
return 0, err
}
var count int64
err = gdb.Table("archived_events").Count(&count).Error
if err != nil {
return 0, err
}
return count, nil
}
// removeArchiveFiles simulates an operator moving the archive
// away by deleting the SQLite file and its sidecar files.
func removeArchiveFiles(t *testing.T, path string) {
t.Helper()
for _, suffix := range []string{
"", "-wal", "-shm", "-journal",
} {
err := os.Remove(path + suffix)
if err != nil && !os.IsNotExist(err) {
t.Fatalf("removing %s%s: %v", path, suffix, err)
}
}
// archivedRow is the columns of an archive row these tests read.
type archivedRow struct {
EventID string
WebhookID string
Method string
RawQuery string
Body string
}
// TestDeliverDatabase_ArchivesEvent verifies that delivering to
@@ -107,9 +244,9 @@ func TestDeliverDatabase_ArchivesEvent(t *testing.T) {
rdb := openArchiveDBForRead(t, archivePath)
var rows []delivery.ExportArchivedEvent
var rows []archivedRow
require.NoError(t, rdb.Find(&rows).Error)
require.NoError(t, rdb.Table("archived_events").Find(&rows).Error)
require.Len(t, rows, 1)
assert.Equal(t, event.ID, rows[0].EventID)
assert.Equal(t, event.WebhookID, rows[0].WebhookID)
@@ -118,262 +255,6 @@ func TestDeliverDatabase_ArchivesEvent(t *testing.T) {
assert.JSONEq(t, `{"archived":true}`, rows[0].Body)
}
func TestArchiveWriter_WritesRow(t *testing.T) {
t.Parallel()
path := filepath.Join(t.TempDir(), "archive-wh.db")
w := delivery.NewExportArchiveWriter(
path, archiveTestLogger(), 0,
)
row := delivery.ExportArchivedEvent{
EventID: "ev-1",
WebhookID: "wh-1",
EntrypointID: "ep-1",
Method: "POST",
Headers: `{"X":"Y"}`,
Body: `{"hello":"world"}`,
ContentType: "application/json",
}
require.NoError(t, w.Write(row, 0))
assert.FileExists(t, path)
var got []delivery.ExportArchivedEvent
require.NoError(t, w.DB().Find(&got).Error)
require.Len(t, got, 1)
assert.Equal(t, "ev-1", got[0].EventID)
assert.Equal(t, "wh-1", got[0].WebhookID)
assert.Equal(t, "ep-1", got[0].EntrypointID)
assert.Equal(t, row.Method, got[0].Method)
assert.Equal(t, row.ContentType, got[0].ContentType)
assert.JSONEq(t, `{"hello":"world"}`, got[0].Body)
assert.False(t, got[0].ArchivedAt.IsZero())
var archivedAt string
require.NoError(t, w.DB().Raw(
"SELECT CAST(archived_at AS TEXT) FROM archived_events",
).Row().Scan(&archivedAt))
assertStoredTime(t, archivedAt)
}
func TestArchiveWriter_RecreatesAfterRemoval(
t *testing.T,
) {
t.Parallel()
path := filepath.Join(t.TempDir(), "archive-wh.db")
w := delivery.NewExportArchiveWriter(
path, archiveTestLogger(), 0,
)
require.NoError(t, w.Write(
delivery.ExportArchivedEvent{EventID: "a"}, 0,
))
assert.FileExists(t, path)
// The operator moves the archive away while the handle is
// still open.
removeArchiveFiles(t, path)
require.NoFileExists(t, path)
// The next write recreates the file with a fresh schema and
// only the new row.
require.NoError(t, w.Write(
delivery.ExportArchivedEvent{EventID: "b"}, 0,
))
assert.FileExists(t, path)
var got []delivery.ExportArchivedEvent
require.NoError(t, w.DB().Find(&got).Error)
require.Len(t, got, 1)
assert.Equal(t, "b", got[0].EventID)
}
// TestStatArchive proves StatArchive finds no file before the first
// write; after a write still held in the -wal, counts the -wal in the
// size and takes its later time as the last write; and finds no file
// again once the file has been moved away.
func TestStatArchive(t *testing.T) {
t.Parallel()
path := filepath.Join(t.TempDir(), "archive-wh.db")
_, err := delivery.StatArchive(path)
require.ErrorIs(t, err, fs.ErrNotExist)
// With the clock stopped, the reopen debounce never passes, so
// the handle stays open after the write.
stopped := time.Now()
w := delivery.NewExportArchiveWriter(path, archiveTestLogger(), 0)
w.SetNow(func() time.Time { return stopped })
require.NoError(t, w.Write(delivery.ExportArchivedEvent{EventID: "a"}, 0))
written := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
earlier := written.Add(-time.Hour)
require.NoError(t, os.Chtimes(path, earlier, earlier))
require.NoError(t, os.Chtimes(path+"-wal", written, written))
file, err := os.Stat(path)
require.NoError(t, err)
wal, err := os.Stat(path + "-wal")
require.NoError(t, err)
require.Positive(t, wal.Size())
got, err := delivery.StatArchive(path)
require.NoError(t, err)
assert.Equal(t, 1, got.Files)
assert.Equal(t, file.Size()+wal.Size(), got.Size)
assert.True(t, written.Equal(got.Written), got.Written)
removeArchiveFiles(t, path)
_, err = delivery.StatArchive(path)
require.ErrorIs(t, err, fs.ErrNotExist)
}
func TestArchiveWriter_ReopenDebounce(t *testing.T) {
t.Parallel()
const debounce = 2 * time.Second
path := filepath.Join(t.TempDir(), "archive-wh.db")
w := delivery.NewExportArchiveWriter(
path, archiveTestLogger(), debounce,
)
// The writer measures its reopen debounce on this clock, which
// only the test moves, so how long the host takes between
// writes cannot change the result.
now := time.Now()
w.SetNow(func() time.Time { return now })
require.NoError(t, w.Write(
delivery.ExportArchivedEvent{EventID: "a"}, 0,
))
require.NoError(t, w.Write(
delivery.ExportArchivedEvent{EventID: "b"}, 0,
))
// Two writes inside the debounce window trigger only the
// initial open — no extra close/reopen.
assert.Equal(t, 1, w.Reopens())
now = now.Add(debounce)
require.NoError(t, w.Write(
delivery.ExportArchivedEvent{EventID: "c"}, 0,
))
// A write after the window elapses closes and reopens once.
assert.Equal(t, 2, w.Reopens())
}
func TestArchiveWriter_ExpiryPrune(t *testing.T) {
t.Parallel()
path := filepath.Join(t.TempDir(), "archive-wh.db")
w := delivery.NewExportArchiveWriter(
path, archiveTestLogger(), 0,
)
require.NoError(t, w.Open(0))
old := delivery.ExportArchivedEvent{
EventID: "old",
ArchivedAt: time.Now().Add(-2 * time.Hour),
}
fresh := delivery.ExportArchivedEvent{
EventID: "fresh",
ArchivedAt: time.Now(),
}
require.NoError(t, w.DB().Create(&old).Error)
require.NoError(t, w.DB().Create(&fresh).Error)
// Reopening with a one-hour expiry prunes the old row.
require.NoError(t, w.Reopen(time.Hour))
var got []delivery.ExportArchivedEvent
require.NoError(t, w.DB().Find(&got).Error)
require.Len(t, got, 1)
assert.Equal(t, "fresh", got[0].EventID)
}
func TestParseArchiveExpiry(t *testing.T) {
t.Parallel()
cases := []struct {
name string
in string
want time.Duration
wantErr bool
}{
{"empty config", "", 0, false},
{"explicit never", `{"expiry":"never"}`, 0, false},
{"empty expiry", `{"expiry":""}`, 0, false},
{"duration", `{"expiry":"1h"}`, time.Hour, false},
{"unparseable", `{"expiry":"nonsense"}`, 0, true},
{"zero duration", `{"expiry":"0s"}`, 0, true},
{"negative duration", `{"expiry":"-5h"}`, 0, true},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
got, err := delivery.ExportParseArchiveExpiry(tc.in)
if tc.wantErr {
require.Error(t, err)
return
}
require.NoError(t, err)
assert.Equal(t, tc.want, got)
})
}
}
// seedDatabaseTargetDelivery seeds a pending delivery of an event
// to a database target and returns the in-memory delivery the
// target handler is invoked with.
func seedDatabaseTargetDelivery(
t *testing.T,
webhookDB *gorm.DB,
event database.Event,
tgt *database.Target,
) *database.Delivery {
t.Helper()
dlv := seedDelivery(
t, webhookDB, event.ID, tgt.ID,
database.DeliveryStatusPending,
)
d := &database.Delivery{
EventID: event.ID,
TargetID: tgt.ID,
Status: database.DeliveryStatusPending,
Event: event,
Target: database.Target{
Name: tgt.Name,
Type: database.TargetTypeDatabase,
Config: tgt.Config,
},
}
d.ID = dlv.ID
return d
}
// TestDeliverDatabase_ArchiveFailureFailsDelivery verifies that
// an archive error (here: an unparseable expiry in the target
// config) fails the delivery loudly: the attempt is recorded as
@@ -423,348 +304,3 @@ func TestDeliverDatabase_ArchiveFailureFailsDelivery(
"no archive file should exist for a failed config",
)
}
func TestValidateArchiveExpiry(t *testing.T) {
t.Parallel()
valid := []string{"", "never", "1h", "720h", "30m"}
for _, in := range valid {
require.NoError(t,
delivery.ValidateArchiveExpiry(in),
"expiry %q should be accepted", in,
)
}
invalid := []string{"nonsense", "7d", "-5h", "0s", "0"}
for _, in := range invalid {
require.Error(t,
delivery.ValidateArchiveExpiry(in),
"expiry %q should be rejected", in,
)
}
}
// TestArchiveFileName pins the archive file name and the rules
// that make a webhook or target name safe to put in it.
func TestArchiveFileName(t *testing.T) {
t.Parallel()
const id = "3f2a1c9e-8d4b-4c1a-9e2f-0a1b2c3d4e5f"
cases := []struct {
name string
webhook string
target string
want string
}{
{
"plain names", "orders", "archive",
"archive-orders-archive-" + id + ".db",
},
{
"lowercased", "Orders", "Main Archive",
"archive-orders-main-archive-" + id + ".db",
},
{
"a run of other characters is one dash",
`a /\..b`, "c__--d",
"archive-a-b-c-d-" + id + ".db",
},
{
"no dash at either end", " --orders!! ", "(archive)",
"archive-orders-archive-" + id + ".db",
},
{
"path separators", "../../etc/passwd", "a/b",
"archive-etc-passwd-a-b-" + id + ".db",
},
{
"letters outside ASCII are dropped",
"Bestellungen Größe", "café",
"archive-bestellungen-gr-e-caf-" + id + ".db",
},
{
"nothing left is unnamed", "", "!!!",
"archive-unnamed-unnamed-" + id + ".db",
},
{
"cut to 40 characters", strings.Repeat("a", 50), "x",
"archive-" + strings.Repeat("a", 40) + "-x-" + id + ".db",
},
{
"no dash left by the cut",
strings.Repeat("a", 39) + " b", "x",
"archive-" + strings.Repeat("a", 39) + "-x-" + id + ".db",
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
assert.Equal(
t, tc.want,
delivery.ArchiveFileName(tc.webhook, tc.target, id),
)
})
}
}
// TestDeliverDatabase_EachTargetHasItsOwnArchive proves two
// database targets of one webhook archive into separate files.
func TestDeliverDatabase_EachTargetHasItsOwnArchive(t *testing.T) {
t.Parallel()
env := setupArchiveTest(t)
first := env.seedDatabaseTarget(t, "")
second := env.addDatabaseTarget(t, first.WebhookID, "")
webhookDB := testWebhookDB(t)
event := seedEvent(t, webhookDB, `{"n":1}`)
for _, tgt := range []*database.Target{first, second} {
env.eng.ExportDeliverDatabase(
webhookDB,
seedDatabaseTargetDelivery(t, webhookDB, event, tgt),
)
}
require.NotEqual(
t, env.archivePath(first), env.archivePath(second),
)
assert.Equal(
t, []string{event.ID},
archivedEventIDs(t, env.archivePath(first)),
)
assert.Equal(
t, []string{event.ID},
archivedEventIDs(t, env.archivePath(second)),
)
}
// TestRename_MovesTheFile proves a rename moves the archive, rows
// and all, and that later writes go to the new name.
func TestRename_MovesTheFile(t *testing.T) {
t.Parallel()
env := setupArchiveTest(t)
tgt := env.seedDatabaseTarget(t, "")
oldPath := env.archivePath(tgt)
webhookDB := testWebhookDB(t)
first := seedEvent(t, webhookDB, `{"n":1}`)
env.eng.ExportDeliverDatabase(
webhookDB, seedDatabaseTargetDelivery(t, webhookDB, first, tgt),
)
require.FileExists(t, oldPath)
require.NoError(
t, env.eng.Rename(tgt.ID, "Orders", "Long Term"),
)
newPath := filepath.Join(
env.dataDir, "archive-orders-long-term-"+tgt.ID+".db",
)
assert.NoFileExists(t, oldPath)
assert.Equal(t, []string{first.ID}, archivedEventIDs(t, newPath))
second := seedEvent(t, webhookDB, `{"n":2}`)
env.eng.ExportDeliverDatabase(
webhookDB,
seedDatabaseTargetDelivery(t, webhookDB, second, tgt),
)
assert.ElementsMatch(
t, []string{first.ID, second.ID},
archivedEventIDs(t, newPath),
)
assert.NoFileExists(
t, oldPath, "a write after the rename must use the new name",
)
}
// TestRename_NeverReplacesAFile plants a file at the new name, once
// the .db alone, once a lone -wal and once a lone -shm, and proves
// each time that the rename is refused, the planted file survives,
// and the archive keeps its name and its rows.
func TestRename_NeverReplacesAFile(t *testing.T) {
t.Parallel()
for _, suffix := range archiveFileSuffixes() {
t.Run("planted .db"+suffix, func(t *testing.T) {
t.Parallel()
env := setupArchiveTest(t)
tgt := env.seedDatabaseTarget(t, "")
oldPath := env.archivePath(tgt)
webhookDB := testWebhookDB(t)
first := seedEvent(t, webhookDB, `{"n":1}`)
env.eng.ExportDeliverDatabase(
webhookDB,
seedDatabaseTargetDelivery(t, webhookDB, first, tgt),
)
newPath := filepath.Join(
env.dataDir, "archive-orders-long-term-"+tgt.ID+".db",
)
plantedPath := newPath + suffix
require.NoError(
t, os.WriteFile(plantedPath, []byte("planted"), 0o600),
)
require.ErrorIs(
t, env.eng.Rename(tgt.ID, "Orders", "Long Term"),
delivery.ErrArchiveNameTaken,
)
//nolint:gosec // reads the file the test planted under t.TempDir()
planted, err := os.ReadFile(plantedPath)
require.NoError(t, err)
assert.Equal(t, "planted", string(planted))
second := seedEvent(t, webhookDB, `{"n":2}`)
env.eng.ExportDeliverDatabase(
webhookDB,
seedDatabaseTargetDelivery(t, webhookDB, second, tgt),
)
assert.ElementsMatch(
t, []string{first.ID, second.ID},
archivedEventIDs(t, oldPath),
)
})
}
}
// TestRename_BeforeTheNameIsSaved covers the order the handlers
// use: they rename before they save the new name, so a delivery in
// between must write under the new name although the main database
// still has the old one. It also shows that renaming an archive that
// does not exist yet is not an error.
func TestRename_BeforeTheNameIsSaved(t *testing.T) {
t.Parallel()
env := setupArchiveTest(t)
tgt := env.seedDatabaseTarget(t, "")
require.NoError(
t, env.eng.Rename(tgt.ID, "Orders", "Archive"),
)
webhookDB := testWebhookDB(t)
event := seedEvent(t, webhookDB, `{"n":1}`)
env.eng.ExportDeliverDatabase(
webhookDB, seedDatabaseTargetDelivery(t, webhookDB, event, tgt),
)
assert.FileExists(
t,
filepath.Join(
env.dataDir, "archive-orders-archive-"+tgt.ID+".db",
),
)
assert.NoFileExists(t, env.archivePath(tgt))
}
// TestArchiveWriter_RenameMovesSidecars proves a rename carries
// the -wal and -shm a crash can leave beside an archive no handle
// has opened since. SQLite finds them by name, so a -wal left
// behind would lose the transactions it holds.
func TestArchiveWriter_RenameMovesSidecars(t *testing.T) {
t.Parallel()
dir := t.TempDir()
oldPath := filepath.Join(dir, "archive-old.db")
newPath := filepath.Join(dir, "archive-new.db")
for _, suffix := range archiveFileSuffixes() {
require.NoError(
t, os.WriteFile(oldPath+suffix, []byte(suffix), 0o600),
)
}
w := delivery.NewExportArchiveWriter(
oldPath, archiveTestLogger(), 0,
)
require.NoError(t, w.Rename("archive-new.db"))
for _, suffix := range archiveFileSuffixes() {
assert.NoFileExists(t, oldPath+suffix)
assert.FileExists(t, newPath+suffix)
}
assert.Equal(t, newPath, w.Path())
}
// TestArchiveWriter_RenameMovesBackOnFailure makes the -wal fail to
// move after the .db has moved, and proves the .db is moved back, so
// the archive is never split across two names. The new name is 255
// bytes, the longest a file name may be, so the .db can take it but
// the -wal, four bytes longer, cannot.
func TestArchiveWriter_RenameMovesBackOnFailure(t *testing.T) {
t.Parallel()
dir := t.TempDir()
oldPath := filepath.Join(dir, "archive-old.db")
newName := strings.Repeat("a", 252) + ".db"
for _, suffix := range archiveFileSuffixes() {
require.NoError(
t, os.WriteFile(oldPath+suffix, []byte(suffix), 0o600),
)
}
w := delivery.NewExportArchiveWriter(
oldPath, archiveTestLogger(), 0,
)
require.Error(t, w.Rename(newName))
for _, suffix := range archiveFileSuffixes() {
assert.FileExists(t, oldPath+suffix)
}
assert.NoFileExists(t, filepath.Join(dir, newName))
assert.Equal(t, oldPath, w.Path())
}
// TestArchiveWriter_RenameMovesBackEveryFile renames a target with a
// file without a period and a file for a month, and proves that when
// the month's file fails to move, the file already moved is moved back:
// both files are under the old name with their rows, and nothing is
// under the new name. The new name is 251 bytes, so the file without a
// period, with its -wal and -shm, can take it, but the month's file,
// eight bytes longer, cannot.
func TestArchiveWriter_RenameMovesBackEveryFile(t *testing.T) {
t.Parallel()
dir := t.TempDir()
oldPath := filepath.Join(dir, "archive-old.db")
monthPath := filepath.Join(dir, "archive-old-2026-03.db")
w := delivery.NewExportArchiveWriter(oldPath, archiveTestLogger(), 0)
require.NoError(t, w.WritePeriod(
delivery.ExportArchivedEvent{EventID: "in-none"}, 0, "",
))
require.NoError(t, w.WritePeriod(
delivery.ExportArchivedEvent{EventID: "in-month"}, 0, "2026-03",
))
require.Error(t, w.Rename(strings.Repeat("a", 248)+".db"))
assert.Equal(t, []string{"in-none"}, archivedEventIDs(t, oldPath))
assert.Equal(t, []string{"in-month"}, archivedEventIDs(t, monthPath))
entries, err := os.ReadDir(dir)
require.NoError(t, err)
for _, entry := range entries {
assert.True(t, strings.HasPrefix(entry.Name(), "archive-old"),
"%s is not under the old name", entry.Name())
}
assert.Equal(t, oldPath, w.Path())
}