Move legacy job repair to migration 0011 and keep archived dinners pending
R1-A: Migration 0010 is back to its round-2 content, the ALTER TABLE only, since a database that already applied 0010 never runs it again. The repair that writes revision 0 into ground_dinner job payloads without one is now migration 0011, applied by the same boot Migrate call, so before the job runner starts and before FailInterruptedGroundings. TestUpgradeKeepsLegacyQueuedGrounding now upgrades from 0009 and from a database with 0010 already recorded, each with a pending Regular and a queued legacy job; both keep the dinner pending through the boot cleanup and then ground it. R2-A: The boot cleanup no longer fails archived dinners. Archiving stops a running job and leaves the dinner pending, which the cleanup relabelled failed with interrupted. TestRestartKeepsArchivedPendingDinner archives a pending Regular while its job runs and checks the cleanup leaves it pending with no error. Maestro task 5037ff86-181d-448e-927b-c2fd2258a25d. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -226,7 +226,7 @@ automatic** (D-021).
|
|||||||
method, minutes_total or difficulty a member saved while the dinner was
|
method, minutes_total or difficulty a member saved while the dinner was
|
||||||
`pending` is kept; grounding fills only those left empty. At boot, after
|
`pending` is kept; grounding fills only those left empty. At boot, after
|
||||||
the job runner fails the jobs a previous process left running, every
|
the job runner fails the jobs a previous process left running, every
|
||||||
dinner still `pending` with no `ground_dinner` job for its revision
|
unarchived dinner still `pending` with no `ground_dinner` job for its revision
|
||||||
queued or running is set `failed` with `interrupted`, so no dinner stays
|
queued or running is set `failed` with `interrupted`, so no dinner stays
|
||||||
pending with nothing left to ground it.
|
pending with nothing left to ground it.
|
||||||
`GET /api/v1/dinners/{id}/recipe-candidates`
|
`GET /api/v1/dinners/{id}/recipe-candidates`
|
||||||
|
|||||||
@@ -150,6 +150,7 @@ SET grounding_status = 'failed',
|
|||||||
grounding_error = ?1,
|
grounding_error = ?1,
|
||||||
updated_at = ?2
|
updated_at = ?2
|
||||||
WHERE grounding_status = 'pending'
|
WHERE grounding_status = 'pending'
|
||||||
|
AND status <> 'archived'
|
||||||
AND NOT EXISTS (
|
AND NOT EXISTS (
|
||||||
SELECT 1 FROM jobs
|
SELECT 1 FROM jobs
|
||||||
WHERE jobs.kind = ?3
|
WHERE jobs.kind = ?3
|
||||||
@@ -165,10 +166,11 @@ type FailOrphanedPendingDinnerGroundingsParams struct {
|
|||||||
Kind string
|
Kind string
|
||||||
}
|
}
|
||||||
|
|
||||||
// Fails every dinner still pending whose grounding job for its current
|
// Fails every unarchived dinner still pending whose grounding job for its
|
||||||
// grounding_revision is no longer queued or running: the process stopped
|
// current grounding_revision is no longer queued or running: the process
|
||||||
// before the job, or before its handler recorded a result. Run at boot,
|
// stopped before the job, or before its handler recorded a result. An
|
||||||
// after the job runner fails the jobs a previous process left running.
|
// archived dinner is left as it is; archiving stopped its grounding. Run at
|
||||||
|
// boot, after the job runner fails the jobs a previous process left running.
|
||||||
func (q *Queries) FailOrphanedPendingDinnerGroundings(ctx context.Context, arg FailOrphanedPendingDinnerGroundingsParams) (int64, error) {
|
func (q *Queries) FailOrphanedPendingDinnerGroundings(ctx context.Context, arg FailOrphanedPendingDinnerGroundingsParams) (int64, error) {
|
||||||
result, err := q.db.ExecContext(ctx, failOrphanedPendingDinnerGroundings, arg.GroundingError, arg.UpdatedAt, arg.Kind)
|
result, err := q.db.ExecContext(ctx, failOrphanedPendingDinnerGroundings, arg.GroundingError, arg.UpdatedAt, arg.Kind)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -36,10 +36,11 @@ type Querier interface {
|
|||||||
DeleteRecipeIngredients(ctx context.Context, recipeID string) error
|
DeleteRecipeIngredients(ctx context.Context, recipeID string) error
|
||||||
DeleteRecipeSteps(ctx context.Context, recipeID string) error
|
DeleteRecipeSteps(ctx context.Context, recipeID string) error
|
||||||
FailJob(ctx context.Context, arg FailJobParams) (int64, error)
|
FailJob(ctx context.Context, arg FailJobParams) (int64, error)
|
||||||
// Fails every dinner still pending whose grounding job for its current
|
// Fails every unarchived dinner still pending whose grounding job for its
|
||||||
// grounding_revision is no longer queued or running: the process stopped
|
// current grounding_revision is no longer queued or running: the process
|
||||||
// before the job, or before its handler recorded a result. Run at boot,
|
// stopped before the job, or before its handler recorded a result. An
|
||||||
// after the job runner fails the jobs a previous process left running.
|
// archived dinner is left as it is; archiving stopped its grounding. Run at
|
||||||
|
// boot, after the job runner fails the jobs a previous process left running.
|
||||||
FailOrphanedPendingDinnerGroundings(ctx context.Context, arg FailOrphanedPendingDinnerGroundingsParams) (int64, error)
|
FailOrphanedPendingDinnerGroundings(ctx context.Context, arg FailOrphanedPendingDinnerGroundingsParams) (int64, error)
|
||||||
// Records why the background grounding of an unarchived dinner still
|
// Records why the background grounding of an unarchived dinner still
|
||||||
// pending at the grounding_revision it was grounded for failed; a dinner
|
// pending at the grounding_revision it was grounded for failed; a dinner
|
||||||
|
|||||||
@@ -227,10 +227,11 @@ func (s *Service) stopGroundings(dinnerID string, below int64) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// FailInterruptedGroundings fails, with ReasonInterrupted, every dinner
|
// FailInterruptedGroundings fails, with ReasonInterrupted, every unarchived
|
||||||
// still pending whose KindGroundDinner job is no longer queued or running:
|
// dinner still pending whose KindGroundDinner job is no longer queued or
|
||||||
// a previous process stopped before enqueuing the job or before its
|
// running: a previous process stopped before enqueuing the job or before
|
||||||
// handler recorded a result. Call it once at boot, after the job runner's
|
// its handler recorded a result. An archived dinner is left pending, since
|
||||||
|
// archiving, not a restart, stopped its grounding. Call it once at boot, after the job runner's
|
||||||
// Start has failed the jobs that process left running, so no dinner stays
|
// Start has failed the jobs that process left running, so no dinner stays
|
||||||
// pending with nothing left to ground it. It returns how many it failed.
|
// pending with nothing left to ground it. It returns how many it failed.
|
||||||
func (s *Service) FailInterruptedGroundings(ctx context.Context) (int64, error) {
|
func (s *Service) FailInterruptedGroundings(ctx context.Context) (int64, error) {
|
||||||
|
|||||||
@@ -946,12 +946,64 @@ func TestRestartFailsInterruptedGrounding(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestRestartKeepsArchivedPendingDinner: a pending Regular archived while
|
||||||
|
// its job runs stays pending, its job succeeded; the boot cleanup leaves it
|
||||||
|
// as it is rather than failing it as interrupted.
|
||||||
|
func TestRestartKeepsArchivedPendingDinner(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
e := newEnv(t)
|
||||||
|
e.search.Gate = make(chan struct{})
|
||||||
|
c, err := e.svc.CreateRegular(t.Context(), dinners.NewRegular{Name: ptr("Chicken curry")}, member)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
e.waitQuery(t, "Chicken curry")
|
||||||
|
if _, err := e.svc.Archive(t.Context(), c.ID, member); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
job := e.waitJob(t, *c.GroundingJobID)
|
||||||
|
close(e.search.Gate)
|
||||||
|
if job.Status != jobs.StatusSucceeded {
|
||||||
|
t.Fatalf("job = %+v, want succeeded", job)
|
||||||
|
}
|
||||||
|
|
||||||
|
// The database is now as the next boot finds it.
|
||||||
|
n, err := e.svc.FailInterruptedGroundings(t.Context())
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if n != 0 {
|
||||||
|
t.Errorf("failed %d dinners, want 0", n)
|
||||||
|
}
|
||||||
|
if row := e.dinner(t, c.ID); row.Status != dinners.StatusArchived ||
|
||||||
|
row.GroundingStatus != dinners.GroundingPending || row.GroundingError.Valid {
|
||||||
|
t.Errorf("dinner = %s, %s (%s), want archived and pending with no error",
|
||||||
|
row.Status, row.GroundingStatus, row.GroundingError.String)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// TestUpgradeKeepsLegacyQueuedGrounding: a pending Regular whose
|
// TestUpgradeKeepsLegacyQueuedGrounding: a pending Regular whose
|
||||||
// ground_dinner job was queued before migration 0010, so its payload has
|
// ground_dinner job was queued before migration 0010, so its payload has
|
||||||
// no revision, is not failed as interrupted at the first boot after the
|
// no revision, is not failed as interrupted at the first boot after the
|
||||||
// upgrade; its job still grounds it.
|
// upgrade; its job still grounds it. The upgrade starts either from 0009
|
||||||
|
// or from a database that has already applied 0010.
|
||||||
func TestUpgradeKeepsLegacyQueuedGrounding(t *testing.T) {
|
func TestUpgradeKeepsLegacyQueuedGrounding(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
for name, below := range map[string]string{
|
||||||
|
"from 0009": "0010",
|
||||||
|
"from 0010": "0011",
|
||||||
|
} {
|
||||||
|
t.Run(name, func(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
upgradeWithLegacyQueuedGrounding(t, below)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// upgradeWithLegacyQueuedGrounding migrates a database to every migration
|
||||||
|
// named before below, adds a pending Regular with a queued ground_dinner
|
||||||
|
// job whose payload has no revision, then boots on the latest migrations.
|
||||||
|
func upgradeWithLegacyQueuedGrounding(t *testing.T, below string) {
|
||||||
ctx := t.Context()
|
ctx := t.Context()
|
||||||
dataDir := t.TempDir()
|
dataDir := t.TempDir()
|
||||||
st := openStore(t, dataDir)
|
st := openStore(t, dataDir)
|
||||||
@@ -961,7 +1013,7 @@ func TestUpgradeKeepsLegacyQueuedGrounding(t *testing.T) {
|
|||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
for _, name := range files {
|
for _, name := range files {
|
||||||
if name >= "0010" {
|
if name >= below {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
body, err := fs.ReadFile(migrations.FS, name)
|
body, err := fs.ReadFile(migrations.FS, name)
|
||||||
|
|||||||
@@ -175,8 +175,8 @@ func TestMigrateEmbeddedMigrations(t *testing.T) {
|
|||||||
if err := Migrate(context.Background(), s.Writer, migrations.FS); err != nil {
|
if err := Migrate(context.Background(), s.Writer, migrations.FS); err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
if got := versions(migrationRows(t, s)); got != "0000_bootstrap,0001_init,0002_household_settings_updated_by,0003_jobs_status,0004_ai_provider_presets,0005_ai_provider_xai,0006_ai_runs_status,0007_recipe_grounding,0008_dinner_recipe_url,0009_dinner_grounding_error,0010_dinner_grounding_revision" {
|
if got := versions(migrationRows(t, s)); got != "0000_bootstrap,0001_init,0002_household_settings_updated_by,0003_jobs_status,0004_ai_provider_presets,0005_ai_provider_xai,0006_ai_runs_status,0007_recipe_grounding,0008_dinner_recipe_url,0009_dinner_grounding_error,0010_dinner_grounding_revision,0011_ground_dinner_job_revision" {
|
||||||
t.Errorf("versions = %q, want 0000_bootstrap,0001_init,0002_household_settings_updated_by,0003_jobs_status,0004_ai_provider_presets,0005_ai_provider_xai,0006_ai_runs_status,0007_recipe_grounding,0008_dinner_recipe_url,0009_dinner_grounding_error,0010_dinner_grounding_revision", got)
|
t.Errorf("versions = %q, want 0000_bootstrap,0001_init,0002_household_settings_updated_by,0003_jobs_status,0004_ai_provider_presets,0005_ai_provider_xai,0006_ai_runs_status,0007_recipe_grounding,0008_dinner_recipe_url,0009_dinner_grounding_error,0010_dinner_grounding_revision,0011_ground_dinner_job_revision", got)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -196,8 +196,8 @@ func TestMigrateBootstrapThenEmbeddedInitApplies(t *testing.T) {
|
|||||||
if err := Migrate(ctx, s.Writer, migrations.FS); err != nil {
|
if err := Migrate(ctx, s.Writer, migrations.FS); err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
if got := versions(migrationRows(t, s)); got != "0000_bootstrap,0001_init,0002_household_settings_updated_by,0003_jobs_status,0004_ai_provider_presets,0005_ai_provider_xai,0006_ai_runs_status,0007_recipe_grounding,0008_dinner_recipe_url,0009_dinner_grounding_error,0010_dinner_grounding_revision" {
|
if got := versions(migrationRows(t, s)); got != "0000_bootstrap,0001_init,0002_household_settings_updated_by,0003_jobs_status,0004_ai_provider_presets,0005_ai_provider_xai,0006_ai_runs_status,0007_recipe_grounding,0008_dinner_recipe_url,0009_dinner_grounding_error,0010_dinner_grounding_revision,0011_ground_dinner_job_revision" {
|
||||||
t.Errorf("versions = %q, want 0000_bootstrap,0001_init,0002_household_settings_updated_by,0003_jobs_status,0004_ai_provider_presets,0005_ai_provider_xai,0006_ai_runs_status,0007_recipe_grounding,0008_dinner_recipe_url,0009_dinner_grounding_error,0010_dinner_grounding_revision", got)
|
t.Errorf("versions = %q, want 0000_bootstrap,0001_init,0002_household_settings_updated_by,0003_jobs_status,0004_ai_provider_presets,0005_ai_provider_xai,0006_ai_runs_status,0007_recipe_grounding,0008_dinner_recipe_url,0009_dinner_grounding_error,0010_dinner_grounding_revision,0011_ground_dinner_job_revision", got)
|
||||||
}
|
}
|
||||||
if !tableExists(t, s, "dinners") {
|
if !tableExists(t, s, "dinners") {
|
||||||
t.Error("table from 0001_init was not created")
|
t.Error("table from 0001_init was not created")
|
||||||
|
|||||||
@@ -5,12 +5,3 @@
|
|||||||
-- earlier job — queued or running — can no longer write the dinner.
|
-- earlier job — queued or running — can no longer write the dinner.
|
||||||
|
|
||||||
ALTER TABLE dinners ADD COLUMN grounding_revision INTEGER NOT NULL DEFAULT 0;
|
ALTER TABLE dinners ADD COLUMN grounding_revision INTEGER NOT NULL DEFAULT 0;
|
||||||
|
|
||||||
-- Every dinner starts at revision 0, so a ground_dinner job enqueued
|
|
||||||
-- before this migration, whose payload has only dinner_id, is for revision
|
|
||||||
-- 0. Recording it lets the boot cleanup, which matches a pending dinner to
|
|
||||||
-- its job by revision, see a job still queued for its dinner.
|
|
||||||
UPDATE jobs
|
|
||||||
SET payload = json_set(payload, '$.revision', 0)
|
|
||||||
WHERE kind = 'ground_dinner'
|
|
||||||
AND json_type(payload, '$.revision') IS NULL;
|
|
||||||
|
|||||||
@@ -0,0 +1,11 @@
|
|||||||
|
-- 0011_ground_dinner_job_revision: every dinner starts at grounding
|
||||||
|
-- revision 0 (0010), so a ground_dinner job enqueued before 0010, whose
|
||||||
|
-- payload has only dinner_id, is for revision 0. Recording it lets the boot
|
||||||
|
-- cleanup, which matches a pending dinner to its job by revision, see a job
|
||||||
|
-- still queued for its dinner. This is a migration of its own, not part of
|
||||||
|
-- 0010, so it also runs on a database that has already applied 0010.
|
||||||
|
|
||||||
|
UPDATE jobs
|
||||||
|
SET payload = json_set(payload, '$.revision', 0)
|
||||||
|
WHERE kind = 'ground_dinner'
|
||||||
|
AND json_type(payload, '$.revision') IS NULL;
|
||||||
@@ -134,15 +134,17 @@ WHERE id = ?
|
|||||||
AND grounding_revision = ?;
|
AND grounding_revision = ?;
|
||||||
|
|
||||||
-- name: FailOrphanedPendingDinnerGroundings :execrows
|
-- name: FailOrphanedPendingDinnerGroundings :execrows
|
||||||
-- Fails every dinner still pending whose grounding job for its current
|
-- Fails every unarchived dinner still pending whose grounding job for its
|
||||||
-- grounding_revision is no longer queued or running: the process stopped
|
-- current grounding_revision is no longer queued or running: the process
|
||||||
-- before the job, or before its handler recorded a result. Run at boot,
|
-- stopped before the job, or before its handler recorded a result. An
|
||||||
-- after the job runner fails the jobs a previous process left running.
|
-- archived dinner is left as it is; archiving stopped its grounding. Run at
|
||||||
|
-- boot, after the job runner fails the jobs a previous process left running.
|
||||||
UPDATE dinners
|
UPDATE dinners
|
||||||
SET grounding_status = 'failed',
|
SET grounding_status = 'failed',
|
||||||
grounding_error = sqlc.arg(grounding_error),
|
grounding_error = sqlc.arg(grounding_error),
|
||||||
updated_at = sqlc.arg(updated_at)
|
updated_at = sqlc.arg(updated_at)
|
||||||
WHERE grounding_status = 'pending'
|
WHERE grounding_status = 'pending'
|
||||||
|
AND status <> 'archived'
|
||||||
AND NOT EXISTS (
|
AND NOT EXISTS (
|
||||||
SELECT 1 FROM jobs
|
SELECT 1 FROM jobs
|
||||||
WHERE jobs.kind = sqlc.arg(kind)
|
WHERE jobs.kind = sqlc.arg(kind)
|
||||||
|
|||||||
Reference in New Issue
Block a user