diff --git a/modules/conga/commands.go b/modules/conga/commands.go index 399d3bf3..baf04a76 100644 --- a/modules/conga/commands.go +++ b/modules/conga/commands.go @@ -120,7 +120,7 @@ asks first; --force skips the question. Off a terminal it never asks.`, if err := confirmProduction(app, in, out, fmt.Sprintf("Delete every pending job of queue %q in production?", queue)); err != nil { return err } - return withDB(ctx, app, func() error { + return withAppDB(ctx, app, func() error { return clearQueue(ctx, app, queue, out) }) }, @@ -336,7 +336,9 @@ func dueAt(c pact.Cadence, t time.Time) bool { } // withAppDB runs fn with the app's database: the published one when there is -// one, otherwise one opened for the command the way withDB does. +// one, otherwise one opened for the command the way withDB does. It serves +// schedule:run --once and queue:clear, which may run in-process on an app +// that already published its pool. func withAppDB(ctx context.Context, app *backpack.App, fn func() error) error { if gdb, ok := app.Lookup[*gorm.DB](); ok && gdb != nil { return fn() diff --git a/modules/conga/commands_test.go b/modules/conga/commands_test.go index c3590792..fd4e90f4 100644 --- a/modules/conga/commands_test.go +++ b/modules/conga/commands_test.go @@ -19,7 +19,7 @@ import ( // none; a running job and other queues are left alone. func TestQueueClear(t *testing.T) { db, dsn := migratedDB(t) - app, gdb := testApp(t, db, dsn, map[string]any{"queue.queues.default": 1}) + app, gdb := testApp(t, db, dsn, map[string]any{"queue.queues.default": 1, "app.key": testAppKey}) m, err := From(app) if err != nil { t.Fatal(err) @@ -73,6 +73,21 @@ func TestQueueClear(t *testing.T) { if !strings.Contains(out, `Clearing queue "default"`) || !strings.Contains(out, "Cleared 2 jobs") { t.Fatalf("output = %q", out) } + // In process, on the app that already published its pool (a plugin + // HasSchedule entry naming queue:clear): the block still holds the one + // worker slot, so these two stay pending. + for range 2 { + if err := m.Enqueue(ctx, nil, pingArgs{Tag: "pending-in-process"}, EnqueueOpts{}); err != nil { + t.Fatal(err) + } + } + var inProcess bytes.Buffer + if err := bonfire.Call(ctx, RuntimeCommands(app, nil), "queue:clear", nil, &inProcess); err != nil { + t.Fatalf("in-process queue:clear: %v\n%s", err, inProcess.String()) + } + if !strings.Contains(inProcess.String(), "Cleared 2 jobs") { + t.Fatalf("in-process output = %q", inProcess.String()) + } close(release) deadline := time.Now().Add(10 * time.Second) for countRows(t, gdb, `SELECT count(*) FROM river_job WHERE kind = ? AND state = 'completed'`, "conga_test_block") != 1 {