clawbot
926d34b4b8
check / check (push) Waiting to run
Each webhook's event database keeps running totals: one row for its events, with when the newest arrived, and one row per target for its deliveries, delivered and failed, each with what retention removed. Every write to them shares the transaction of the rows it counts, and a delivery already delivered or failed is not settled again. Deliveries get a finished_at column; it and target_id end the status index, so each target's deliveries finished in a window come from one index-range query grouped by target. Retention deletes 1000 expired events per transaction, pausing 200 ms between them so other writers get in, and stops between them on shutdown. The pane is its own template, its figures in tables. The schema changes in place with nothing back-filled, so an existing database must be recreated. Model: opus-5-5
98 lines
2.6 KiB
Go
98 lines
2.6 KiB
Go
package database
|
|
|
|
import (
|
|
"context"
|
|
"io"
|
|
"log/slog"
|
|
"os"
|
|
"testing"
|
|
"time"
|
|
|
|
"go.uber.org/fx"
|
|
)
|
|
|
|
// NewTestRetentionReaper builds a RetentionReaper backed by the given
|
|
// main database and per-webhook database manager, without the fx
|
|
// lifecycle. Intended for tests.
|
|
func NewTestRetentionReaper(
|
|
db *Database,
|
|
mgr *WebhookDBManager,
|
|
) *RetentionReaper {
|
|
return &RetentionReaper{
|
|
db: db,
|
|
dbManager: mgr,
|
|
log: slog.New(slog.NewTextHandler(
|
|
os.Stderr,
|
|
&slog.HandlerOptions{Level: slog.LevelDebug},
|
|
)),
|
|
interval: time.Hour,
|
|
}
|
|
}
|
|
|
|
// ExportReapBatchSize exposes how many expired events one retention
|
|
// transaction deletes.
|
|
const ExportReapBatchSize = reapBatchSize
|
|
|
|
// ExportSweep runs a single retention sweep synchronously for tests.
|
|
func (r *RetentionReaper) ExportSweep(ctx context.Context) {
|
|
r.sweep(ctx)
|
|
}
|
|
|
|
// ExportRegisterHooks registers the reaper'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 (r *RetentionReaper) ExportRegisterHooks(lc fx.Lifecycle) {
|
|
r.registerHooks(lc)
|
|
}
|
|
|
|
// ExportStart starts the reaper's background loop for tests.
|
|
func (r *RetentionReaper) ExportStart() {
|
|
r.start()
|
|
}
|
|
|
|
// ExportStop stops the reaper's background loop for tests.
|
|
func (r *RetentionReaper) ExportStop(ctx context.Context) error {
|
|
return r.stop(ctx)
|
|
}
|
|
|
|
// ExportWedgeLoop adds a goroutine to the reaper's WaitGroup that
|
|
// never observes cancellation and returns only when release is
|
|
// closed. It stands in for a sweep stuck on a locked database.
|
|
func (r *RetentionReaper) ExportWedgeLoop(
|
|
release <-chan struct{},
|
|
) {
|
|
r.wg.Go(func() {
|
|
<-release
|
|
})
|
|
}
|
|
|
|
// ExportSetInterval overrides the sweep interval for tests.
|
|
func (r *RetentionReaper) ExportSetInterval(d time.Duration) {
|
|
r.interval = d
|
|
}
|
|
|
|
// ExportSetBannerOut redirects the first-boot credentials banner, so a
|
|
// test can read what the operator would have seen. It must be called
|
|
// before the fx start hook runs, which is where the account is seeded.
|
|
func (d *Database) ExportSetBannerOut(w io.Writer) {
|
|
d.bannerOut = w
|
|
}
|
|
|
|
// DummyPasswordHashForTest exposes the encoded hash that unknown
|
|
// usernames are verified against.
|
|
func DummyPasswordHashForTest() string {
|
|
return dummyPasswordHash()
|
|
}
|
|
|
|
// HashAtShippedCostForTest makes HashPassword hash at the shipped
|
|
// memory cost until t ends. t must not run in parallel with other
|
|
// tests, which would hash at that cost alongside it.
|
|
func HashAtShippedCostForTest(t *testing.T) {
|
|
t.Helper()
|
|
|
|
hashAtShippedCostInTest = true
|
|
|
|
t.Cleanup(func() { hashAtShippedCostInTest = false })
|
|
}
|