From 56dd5599cca09b6de21a0d7fe540b0ec9b2dc32d Mon Sep 17 00:00:00 2001 From: overseer Date: Sat, 8 Aug 2026 16:40:16 +0200 Subject: [PATCH] overseer: exec iteration 1 --- internal/corrosion/vms.go | 54 +++++++++++++++++++++--------- internal/corrosion/vms_test.go | 33 ++++++++++++++++++ internal/grpcapi/ha_health_test.go | 27 ++++++++++++++- 3 files changed, 97 insertions(+), 17 deletions(-) diff --git a/internal/corrosion/vms.go b/internal/corrosion/vms.go index 3c2352dd..5518cfe5 100644 --- a/internal/corrosion/vms.go +++ b/internal/corrosion/vms.go @@ -58,7 +58,9 @@ type VMRecord struct { // protocol columns: OwnerEpoch bumps on every ownership transfer (ABA-proof // recovery), SpecGeneration bumps on every desired-spec mutation, and // ActiveOperationID is the VM-wide mutation barrier (non-empty ⇒ an operation - // holds the VM). Populated by GetVM; the ListVMs projection omits them. + // holds the VM). All three are populated by GetVM; SpecGeneration and + // ActiveOperationID are omitted by BOTH list projections, while OwnerEpoch + // rides the unpaginated ListVMs read only (not ListVMsPage) — see scanVMRow. OwnerEpoch int64 SpecGeneration int64 ActiveOperationID string @@ -248,6 +250,7 @@ func ListVMs(ctx context.Context, c *Client, stackName, hostName string) ([]VMRe sql := `SELECT name, stack_name, host_name, spec, state, state_detail, cpu_actual, mem_actual, COALESCE(project, '_default') AS project, COALESCE(is_template, 0) AS is_template, + COALESCE(pending_action_id, '') AS pending_action_id, COALESCE(vm_owner_epoch, 0) AS vm_owner_epoch, created_at, updated_at FROM vms WHERE deleted_at IS NULL` var params []interface{} @@ -273,27 +276,45 @@ func ListVMs(ctx context.Context, c *Client, stackName, hostName string) ([]VMRe return vms, nil } -// scanVMRow maps a row carrying the ListVMs column set to a VMRecord. +// scanVMRow maps a row from a VM-list projection to a VMRecord. It reads the UNION of the +// columns its callers select, and an absent column reads as a ZERO VALUE, not an error +// (Row.String/Int64 on a missing column) — so a field is only trustworthy on the paths whose +// SELECT actually carries it: +// +// - pending_action_id: carried by BOTH ListVMs and ListVMsPage. It must be, because +// grpcapi's anyStrandedPending treats an empty marker on a pending VM as a stranded +// transfer — an omission here reads as "markerless" and reports a legitimately-minted +// transfer as stranded (fixed here; previously omitted by both). +// - vm_owner_epoch: carried by ListVMs ONLY, so OwnerEpoch reads 0 through ListVMsPage. +// Deliberate: the dual-run detector's index is built from the unpaginated ListVMs and no +// ListVMsPage consumer reads OwnerEpoch. Anything that starts reading it on the +// paginated path must add the column there first. +// - spec_generation / active_operation_id: read by NEITHER list projection. Use GetVM or +// ListVMsWithActiveOperation. +// +// Adding a field to this scanner therefore means adding its column to every caller whose +// consumers actually read that field. func scanVMRow(r Row) VMRecord { return VMRecord{ - Name: r.String("name"), - StackName: r.String("stack_name"), - HostName: r.String("host_name"), - Spec: r.String("spec"), - State: r.String("state"), - StateDetail: r.String("state_detail"), - CPUActual: r.Int("cpu_actual"), - MemActual: r.Int("mem_actual"), - Project: r.String("project"), - IsTemplate: r.Int("is_template") == 1, + Name: r.String("name"), + StackName: r.String("stack_name"), + HostName: r.String("host_name"), + Spec: r.String("spec"), + State: r.String("state"), + StateDetail: r.String("state_detail"), + CPUActual: r.Int("cpu_actual"), + MemActual: r.Int("mem_actual"), + Project: r.String("project"), + IsTemplate: r.Int("is_template") == 1, + PendingActionID: r.String("pending_action_id"), // OwnerEpoch rides the list read because the dual-run detector's DB // index is built from ListVMs. Omitting it made every epoched running // VM read as marker-vs-0 and page a false owner_epoch_mismatch — a bug // the LAB caught, not the unit tests: the fixture VMs happened to be // epoch 0, so marker 0 == "missing epoch" 0 and nothing fired. - OwnerEpoch: r.Int64("vm_owner_epoch"), - CreatedAt: r.String("created_at"), - UpdatedAt: r.String("updated_at"), + OwnerEpoch: r.Int64("vm_owner_epoch"), + CreatedAt: r.String("created_at"), + UpdatedAt: r.String("updated_at"), } } @@ -304,7 +325,8 @@ func scanVMRow(r Row) VMRecord { func ListVMsPage(ctx context.Context, c *Client, stackName, hostName, afterName string, limit int) ([]VMRecord, error) { sql := `SELECT name, stack_name, host_name, spec, state, state_detail, cpu_actual, mem_actual, COALESCE(project, '_default') AS project, - COALESCE(is_template, 0) AS is_template, created_at, updated_at + COALESCE(is_template, 0) AS is_template, + COALESCE(pending_action_id, '') AS pending_action_id, created_at, updated_at FROM vms WHERE deleted_at IS NULL` var params []interface{} if stackName != "" { diff --git a/internal/corrosion/vms_test.go b/internal/corrosion/vms_test.go index 94c4d7f4..d01f54e3 100644 --- a/internal/corrosion/vms_test.go +++ b/internal/corrosion/vms_test.go @@ -191,6 +191,39 @@ func TestListVMs_Filter(t *testing.T) { } } +// pending_action_id must be in BOTH list projections, because an absent column reads as "" +// rather than erroring and grpcapi's anyStrandedPending treats an empty marker on a pending +// VM as a stranded transfer — so an omission on either path reports a legitimately-minted +// transfer as stranded. (This is a per-column requirement, not a claim that the two +// projections are otherwise identical: vm_owner_epoch is intentionally ListVMs-only. See +// scanVMRow's comment for the full per-column breakdown.) +func TestListVMs_SurfacesPendingActionID(t *testing.T) { + ctx := context.Background() + c := apTestClient(t) + apInsertVM(t, c, "vm1", "h1", "running") + if err := WriteVMRescheduleProof(ctx, c, apProof("p1", "vm1", "h1"), "vm1", "h1"); err != nil { + t.Fatalf("WriteVMRescheduleProof: %v", err) + } + for _, tc := range []struct { + name string + list func() ([]VMRecord, error) + }{ + {"ListVMs", func() ([]VMRecord, error) { return ListVMs(ctx, c, "", "h1") }}, + {"ListVMsPage", func() ([]VMRecord, error) { return ListVMsPage(ctx, c, "", "h1", "", 10) }}, + } { + vms, err := tc.list() + if err != nil || len(vms) != 1 { + t.Fatalf("%s: err=%v rows=%d; want 1 row", tc.name, err, len(vms)) + } + if vms[0].State != "pending" { + t.Fatalf("%s: state=%q; want pending", tc.name, vms[0].State) + } + if vms[0].PendingActionID != "p1" { + t.Fatalf("%s: PendingActionID=%q; want p1", tc.name, vms[0].PendingActionID) + } + } +} + func TestUpdateVMState(t *testing.T) { c, err := NewTestClient() if err != nil { diff --git a/internal/grpcapi/ha_health_test.go b/internal/grpcapi/ha_health_test.go index 4c081f3d..80c19b1b 100644 --- a/internal/grpcapi/ha_health_test.go +++ b/internal/grpcapi/ha_health_test.go @@ -82,7 +82,32 @@ func TestAnyStrandedPending(t *testing.T) { if s.anyStrandedPending(ctx) { t.Fatal("a running VM is not a stranded pending transfer") } - // A markerless state=pending VM assigned here → stranded. + // A state=pending VM that DOES carry a pending_action_id is a legitimately minted + // transfer, not stranded — the marker is exactly what startPendingVM validates. + // This case was unreachable while ListVMs' projection omitted the column: every + // pending VM read as markerless, so a routine reschedule onto this host flapped + // legacy_pending_stranded. + if err := corrosion.InsertVM(ctx, s.db, corrosion.VMRecord{ + Name: "vmok", HostName: "test-host", State: "stopped", + }, nil, nil); err != nil { + t.Fatalf("InsertVM: %v", err) + } + // Mint the pending transition the way the failover coordinator does: proof row + + // host/state/pending_action_id stamped in ONE batch. + if err := corrosion.WriteVMRescheduleProof(ctx, s.db, corrosion.ActionProof{ + ID: "p-ok", Action: corrosion.ActionReschedule, TargetKind: "vm", + TargetName: "vmok", DestHost: "test-host", Coordinator: "coord-1", + LeaseHolder: "coord-1", QuorumLive: 3, QuorumNeeded: 2, + }, "vmok", "test-host"); err != nil { + t.Fatalf("WriteVMRescheduleProof: %v", err) + } + if s.anyStrandedPending(ctx) { + t.Fatal("a pending VM carrying a pending_action_id is a minted transfer, not stranded") + } + // A markerless state=pending VM assigned here → stranded. Asserted LAST on purpose: + // anyStrandedPending is an OR over rows, so the negative case above has to run while + // no stranded row exists. Landing here with vmok already in the table also proves the + // predicate is per-row — neither "any pending ⇒ true" nor "any marker ⇒ false". if err := corrosion.InsertVM(ctx, s.db, corrosion.VMRecord{ Name: "vmbad", HostName: "test-host", State: "pending", }, nil, nil); err != nil {