1 Commits
Author SHA1 Message Date
sneak ab7eec4b3e Move the archive files into their own package (closes #541)
check / check (push) Canceled after 0s
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
2026-10-15 18:42:29 +00:00
11 changed files with 97 additions and 37 deletions
+2 -2
View File
@@ -189,8 +189,8 @@ func newApp() *fx.App {
fx.Provide( fx.Provide(
globals.New, globals.New,
logger.New, logger.New,
// The plain logger the session, the middleware and the // The plain logger the session, the middleware, the
// webhook database manager take. // webhook database manager and the archive writers take.
func(l *logger.Logger) *slog.Logger { return l.Get() }, func(l *logger.Logger) *slog.Logger { return l.Get() },
config.New, config.New,
database.New, database.New,
+9 -13
View File
@@ -26,8 +26,8 @@ const archiveNameMaxLen = 40
// target releases the cached archive writers, whose open file // target releases the cached archive writers, whose open file
// handles would otherwise outlive them; renaming one renames the // handles would otherwise outlive them; renaming one renames the
// archive files, which are named for the webhook and the target // archive files, which are named for the webhook and the target
// (see FileName). Lock and Unlock take and release the lock every // (see FileName). RenameLock returns the lock every rename runs under
// rename runs under (see Writers.Lock). // (see Writers.RenameLock).
// //
// It is an interface so that the handlers' tests can stand in for // It is an interface so that the handlers' tests can stand in for
// Writers. // Writers.
@@ -35,7 +35,7 @@ const archiveNameMaxLen = 40
// Neither eviction deletes an archive file. Both are idempotent // Neither eviction deletes an archive file. Both are idempotent
// and are no-ops for a webhook or target with no writer. // and are no-ops for a webhook or target with no writer.
type Archives interface { type Archives interface {
sync.Locker RenameLock() *sync.Mutex
EvictWebhook(webhookID string) EvictWebhook(webhookID string)
EvictTarget(targetID string) EvictTarget(targetID string)
Rename(targetID, webhookName, targetName string) error Rename(targetID, webhookName, targetName string) error
@@ -74,7 +74,7 @@ type Writers struct {
// stored names from the other. An archive download holds it while // stored names from the other. An archive download holds it while
// it reads the stored names and lists the files they give, and // it reads the stored names and lists the files they give, and
// again for each file while it finds the file under the names // again for each file while it finds the file under the names
// stored then and opens it. Lock and Unlock take and release it. // stored then and opens it. RenameLock returns it.
renameMu sync.Mutex renameMu sync.Mutex
} }
@@ -88,15 +88,11 @@ func NewWriters(params WritersParams) *Writers {
} }
} }
// Lock takes the lock every rename of archive files runs under: see // RenameLock implements Archives. It returns the lock every rename of
// renameMu for who holds it and for how long. // archive files runs under: see renameMu for who holds it and for how
func (ws *Writers) Lock() { // long. Archive writes and Rename itself do not take it.
ws.renameMu.Lock() func (ws *Writers) RenameLock() *sync.Mutex {
} return &ws.renameMu
// Unlock releases the lock Lock takes.
func (ws *Writers) Unlock() {
ws.renameMu.Unlock()
} }
// FileName returns the file name of a database target's // FileName returns the file name of a database target's
+60
View File
@@ -711,3 +711,63 @@ func TestArchiveWriter_RenameMovesBackEveryFile(t *testing.T) {
assert.Equal(t, oldPath, w.Path()) assert.Equal(t, oldPath, w.Path())
} }
// TestRenameLock_HeldByOneEditAtATime holds the rename lock, as an edit
// does from loading the stored names until it has saved them. While it
// is held, the edit's rename and an archive write still finish, since
// neither takes it, and the event ends up under the new name whichever
// ran first. A second edit waits for the lock until the first lets go.
func TestRenameLock_HeldByOneEditAtATime(t *testing.T) {
t.Parallel()
env := setupArchiveTest(t)
tgt := env.seedDatabaseTarget(t, "")
event := newEvent(`{"n":1}`)
env.writers.RenameLock().Lock()
finished := make(chan error, 2)
go func() {
finished <- env.writers.Rename(tgt.ID, "Orders", "Long Term")
}()
go func() {
finished <- env.writers.Write(databaseTargetDelivery(event, tgt))
}()
for range 2 {
select {
case err := <-finished:
require.NoError(t, err)
case <-time.After(10 * time.Second):
t.Fatal("a rename or a write waited for the rename lock")
}
}
newPath := filepath.Join(
env.dataDir, "archive-orders-long-term-"+tgt.ID+".db",
)
assert.Equal(t, []string{event.ID}, archivedEventIDs(t, newPath))
secondEdit := make(chan struct{})
go func() {
env.writers.RenameLock().Lock()
close(secondEdit)
env.writers.RenameLock().Unlock()
}()
select {
case <-secondEdit:
t.Fatal("a second edit took the rename lock while it was held")
case <-time.After(100 * time.Millisecond):
}
env.writers.RenameLock().Unlock()
select {
case <-secondEdit:
case <-time.After(10 * time.Second):
t.Fatal("a second edit did not get the rename lock once released")
}
}
+1 -1
View File
@@ -106,7 +106,7 @@ type exportedName struct {
// NewExport lists a database target's archive files for export, // NewExport lists a database target's archive files for export,
// given the path Path gives it (see archiveFiles). It opens none // given the path Path gives it (see archiveFiles). It opens none
// of them. Its caller holds lock, which every rename of the target's // of them. Its caller holds lock, which every rename of the target's
// files runs under (Writers.Lock takes it), from reading the names // files runs under (Writers.RenameLock returns it), from reading the names
// path is made of until it returns, so the files it lists are the ones // path is made of until it returns, so the files it lists are the ones
// those names give. // those names give.
// //
+3 -7
View File
@@ -61,7 +61,7 @@ func (n *recordingNotifier) Tasks() []delivery.Task {
// path reached the archive writers. After FailRenames, every // path reached the archive writers. After FailRenames, every
// rename of that target fails with the given error. After // rename of that target fails with the given error. After
// BlockNextRename, the next rename is recorded and then waits. // BlockNextRename, the next rename is recorded and then waits.
// Lock and Unlock are a plain mutex, as in archive.Writers. // RenameLock returns a plain mutex, as in archive.Writers.
type recordingArchives struct { type recordingArchives struct {
renameMu sync.Mutex renameMu sync.Mutex
mu sync.Mutex mu sync.Mutex
@@ -89,12 +89,8 @@ type archiveRename struct {
TargetName string TargetName string
} }
func (r *recordingArchives) Lock() { func (r *recordingArchives) RenameLock() *sync.Mutex {
r.renameMu.Lock() return &r.renameMu
}
func (r *recordingArchives) Unlock() {
r.renameMu.Unlock()
} }
func (r *recordingArchives) EvictWebhook(webhookID string) { func (r *recordingArchives) EvictWebhook(webhookID string) {
+2 -2
View File
@@ -28,8 +28,8 @@ func (h *Handlers) HandleTargetCreate() http.HandlerFunc {
sourceID := chi.URLParam(r, "sourceID") sourceID := chi.URLParam(r, "sourceID")
h.archives.Lock() h.archives.RenameLock().Lock()
defer h.archives.Unlock() defer h.archives.RenameLock().Unlock()
var webhook database.Webhook var webhook database.Webhook
+5 -5
View File
@@ -82,7 +82,7 @@ func (d downloadWriter) Write(b []byte) (int, error) {
// listTargetArchive lists the archive files of the request's database // listTargetArchive lists the archive files of the request's database
// target for export. It reports false once it has written the response. // target for export. It reports false once it has written the response.
// //
// It holds the archives' lock, which every archive rename runs under, // It holds the rename lock, which every archive rename runs under,
// while it reads the stored names and lists the files, so the files it // while it reads the stored names and lists the files, so the files it
// lists are the ones those names give. It lets go before the export is // lists are the ones those names give. It lets go before the export is
// streamed, which takes the lock again for each file only while it // streamed, which takes the lock again for each file only while it
@@ -91,8 +91,8 @@ func (h *Handlers) listTargetArchive(
w http.ResponseWriter, w http.ResponseWriter,
r *http.Request, r *http.Request,
) (database.Webhook, *database.Target, *archive.Export, bool) { ) (database.Webhook, *database.Target, *archive.Export, bool) {
h.archives.Lock() h.archives.RenameLock().Lock()
defer h.archives.Unlock() defer h.archives.RenameLock().Unlock()
webhook, target, ok := h.ownedTarget(w, r) webhook, target, ok := h.ownedTarget(w, r)
if !ok { if !ok {
@@ -107,7 +107,7 @@ func (h *Handlers) listTargetArchive(
export, err := archive.NewExport( export, err := archive.NewExport(
archive.Path(h.dbMgr, &webhook, target), archive.Path(h.dbMgr, &webhook, target),
h.archives, h.archives.RenameLock(),
func() (string, error) { func() (string, error) {
return h.storedArchivePath(webhook.ID, target.ID) return h.storedArchivePath(webhook.ID, target.ID)
}, },
@@ -124,7 +124,7 @@ func (h *Handlers) listTargetArchive(
// storedArchivePath returns the path archive.Path gives a // storedArchivePath returns the path archive.Path gives a
// database target under the names stored for it and its webhook now. // database target under the names stored for it and its webhook now.
// Its caller holds the archives' lock. A webhook or target deleted since is still // Its caller holds the rename lock. A webhook or target deleted since is still
// found, since deleting one leaves its archive files under their names. // found, since deleting one leaves its archive files under their names.
func (h *Handlers) storedArchivePath( func (h *Handlers) storedArchivePath(
webhookID, targetID string, webhookID, targetID string,
+2 -2
View File
@@ -97,8 +97,8 @@ func (h *Handlers) HandleTargetEdit() http.HandlerFunc {
// HandleTargetEditSubmit handles the target edit form submission. // HandleTargetEditSubmit handles the target edit form submission.
func (h *Handlers) HandleTargetEditSubmit() http.HandlerFunc { func (h *Handlers) HandleTargetEditSubmit() http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) { return func(w http.ResponseWriter, r *http.Request) {
h.archives.Lock() h.archives.RenameLock().Lock()
defer h.archives.Unlock() defer h.archives.RenameLock().Unlock()
webhook, target, ok := h.ownedTarget(w, r) webhook, target, ok := h.ownedTarget(w, r)
if !ok { if !ok {
+2 -2
View File
@@ -59,8 +59,8 @@ func (h *Handlers) HandleSourceEditSubmit() http.HandlerFunc {
sourceID := chi.URLParam(r, "sourceID") sourceID := chi.URLParam(r, "sourceID")
h.archives.Lock() h.archives.RenameLock().Lock()
defer h.archives.Unlock() defer h.archives.RenameLock().Unlock()
var webhook database.Webhook var webhook database.Webhook
+5 -1
View File
@@ -136,7 +136,11 @@ type noopNotifier struct{}
func (n *noopNotifier) Notify([]delivery.Task) {} func (n *noopNotifier) Notify([]delivery.Task) {}
type noopArchives struct { type noopArchives struct {
sync.Mutex renameMu sync.Mutex
}
func (n *noopArchives) RenameLock() *sync.Mutex {
return &n.renameMu
} }
func (n *noopArchives) EvictWebhook(string) {} func (n *noopArchives) EvictWebhook(string) {}
+6 -2
View File
@@ -54,9 +54,13 @@ func (n *noopNotifier) Notify([]delivery.Task) {}
// noopArchives satisfies handlers.New's archive.Archives // noopArchives satisfies handlers.New's archive.Archives
// dependency. No test here checks what gets evicted or renamed, so // dependency. No test here checks what gets evicted or renamed, so
// it records nothing. Its lock is a plain mutex. // it records nothing. Its rename lock is a plain mutex.
type noopArchives struct { type noopArchives struct {
sync.Mutex renameMu sync.Mutex
}
func (e *noopArchives) RenameLock() *sync.Mutex {
return &e.renameMu
} }
func (e *noopArchives) EvictWebhook(string) {} func (e *noopArchives) EvictWebhook(string) {}