diff --git a/modules/conga/schedule_test.go b/modules/conga/schedule_test.go index fbc4de46..f0509c13 100644 --- a/modules/conga/schedule_test.go +++ b/modules/conga/schedule_test.go @@ -254,6 +254,44 @@ func TestScheduleRunsCommand(t *testing.T) { logs.waitFor(t, slog.LevelWarn, "schedule: command not registered; skipping", "command", "acme:missing") } +// TestScheduledDeferredPurgeInProcess runs the framework entry deferred:purge +// the way serve's worker does (callScheduled through the published catalog) +// and through schedule:run --once, both on an app whose pool is already +// published. Both used to fail with "backpack: duplicate provider for +// *sql.DB". +func TestScheduledDeferredPurgeInProcess(t *testing.T) { + db, dsn := migratedDB(t) + app, _ := testApp(t, db, dsn, map[string]any{"app.key": testAppKey}) + logs := &captureHandler{} + if err := app.Publish(slog.New(logs)); err != nil { + t.Fatal(err) + } + if err := app.Publish(bonfire.NewCatalog(lagoon.RuntimeCommands(app, nil))); err != nil { + t.Fatal(err) + } + var buf bytes.Buffer + skipped, err := callScheduled(t.Context(), app, loggerFromApp(app), "deferred:purge", nil, &buf) + if skipped || err != nil { + t.Fatalf("callScheduled: skipped=%v err=%v\n%s", skipped, err, buf.String()) + } + if !strings.Contains(buf.String(), "purged 0 deferred bindings older than 5 days") { + t.Fatalf("output = %q", buf.String()) + } + if !logs.find(slog.LevelInfo, "schedule: command finished", "command", "deferred:purge") { + t.Fatal("no Info 'schedule: command finished' record for deferred:purge") + } + if err := db.PingContext(t.Context()); err != nil { + t.Fatalf("published pool closed by the command: %v", err) + } + out, err := onceRun(t, app, nil, mustTime(t, time.UTC, "2026-03-10 03:00:00")) + if err != nil { + t.Fatalf("schedule:run --once: %v\n%s", err, out) + } + if !strings.Contains(out, "Ran 1 due entries (0 failed)") { + t.Fatalf("schedule:run --once output = %q", out) + } +} + // onceRun runs `schedule:run --once` for plugins at the given app time and // returns its output. func onceRun(t *testing.T, app *backpack.App, plugins []party.Plugin, now time.Time) (string, error) { diff --git a/modules/lagoon/commands.go b/modules/lagoon/commands.go index 569c5e97..4099bff7 100644 --- a/modules/lagoon/commands.go +++ b/modules/lagoon/commands.go @@ -483,7 +483,16 @@ func purgeBucket(ctx context.Context, app *backpack.App) (*blob.Bucket, func(), return b, func() { _ = b.Close() }, nil } +// withDB runs fn on the app's published database when there is one: an +// in-process run (the scheduler in serve or schedule:run, through +// bonfire.Catalog.Call) borrows that pool and leaves it open. Otherwise it +// opens and publishes a pool for the CLI run and closes it when fn returns. func withDB(ctx context.Context, app *backpack.App, fn func(*gorm.DB) error) error { + if app != nil { + if gdb, ok := app.Lookup[*gorm.DB](); ok && gdb != nil { + return fn(gdb) + } + } sqlDB, gdb, err := OpenFromApp(ctx, app) if err != nil { return err diff --git a/modules/lagoon/purge_test.go b/modules/lagoon/purge_test.go index 5e1dd163..8a14c651 100644 --- a/modules/lagoon/purge_test.go +++ b/modules/lagoon/purge_test.go @@ -501,3 +501,50 @@ func TestPurgeDeferredCommand(t *testing.T) { t.Fatal("the command left the file, its blob or a binding") } } + +// TestPurgeDeferredCommandPublishedDB runs deferred:purge the way the +// in-process scheduler does: through bonfire.Call on an app whose pool serve +// already opened and published. The command used to publish a second pool +// and fail with "backpack: duplicate provider for *sql.DB"; it now runs on +// the published handle and leaves that pool open. +func TestPurgeDeferredCommandPublishedDB(t *testing.T) { + e := newPurgeEnv(t, "lagoon_purge_published") + f := e.file(t, "") + created := e.child(t, "created") + e.bind(t, "cover", DeferredFileType, fmt.Sprint(f.ID), nil, 3*day) + e.bind(t, "children", "acme_purge_children", fmt.Sprint(created), &DeferredEnvelope{Created: true}, 3*day) + + app := backpack.New(purgeConfig(t, map[string]any{"database.dsn": e.dsn, "app.key": base64.StdEncoding.EncodeToString(bytes.Repeat([]byte{7}, 32))})) + sqlDB, gdb, err := OpenFromApp(t.Context(), app) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = sqlDB.Close() }) + if err := Publish(app, sqlDB, gdb); err != nil { + t.Fatal(err) + } + if err := attach.Publish(app, e.bucket); err != nil { + t.Fatal(err) + } + plugins := []party.Plugin{purgePlugin{models: []any{&purgeChild{}}}} + var out bytes.Buffer + if err := bonfire.Call(t.Context(), RuntimeCommands(app, plugins), "deferred:purge", []string{"--days", "2"}, &out); err != nil { + t.Fatalf("deferred:purge: %v\n%s", err, out.String()) + } + if !strings.Contains(out.String(), "purged 2 deferred bindings older than 2 days (1 files, 1 children, 0 skipped)") { + t.Fatalf("output = %q", out.String()) + } + if e.fileExists(t, f.ID) || e.blobExists(t, f) || len(e.bindingIDs(t)) != 0 { + t.Fatal("the command left the file, its blob or a binding") + } + if err := sqlDB.PingContext(t.Context()); err != nil { + t.Fatalf("published pool closed by the command: %v", err) + } + if got, ok := app.Lookup[*gorm.DB](); !ok || got != gdb { + t.Fatalf("published *gorm.DB replaced: %p, want %p", got, gdb) + } + out.Reset() + if err := bonfire.Call(t.Context(), RuntimeCommands(app, plugins), "migrate:status", nil, &out); err != nil { + t.Fatalf("migrate:status: %v\n%s", err, out.String()) + } +}