diff --git a/cmd/mesh-controller/pending.go b/cmd/mesh-controller/pending.go index 7743f165..977617bb 100644 --- a/cmd/mesh-controller/pending.go +++ b/cmd/mesh-controller/pending.go @@ -491,8 +491,7 @@ func settlingPending(ctx context.Context, open *stores) { // settlePending is one tick: claims left by a controller that stopped are settled from what the mesh holds; // a module registered meanwhile is assigned; a build heard and not registered, or silent past its bound, -// ends its assignment; ended ones are raised as conditions and cleared once answered; ended rows older than -// KeptFor are deleted. Returns what it did. +// ends its assignment; ended ones are raised as conditions and cleared once answered. Returns what it did. func settlePending(ctx context.Context, open *stores, now time.Time) []string { inv := open.inventory var said []string @@ -565,11 +564,6 @@ func settlePending(ctx context.Context, open *stores, now time.Time) []string { } said = append(said, raisePendingEnded(ctx, inv)...) - if n, err := inv.ForgetEndedPending(ctx, now); err != nil { - say("ended pending assignments could not be deleted: %v", err) - } else if n > 0 { - say("deleted %d pending assignment(s) ended more than %s ago", n, inventory.KeptFor) - } return said } @@ -585,7 +579,7 @@ func containsString(xs []string, x string) bool { // pendingObservation is the condition for a pending assignment that was not made: one per pending // assignment, its row's id the last part of its id, so one row's condition is never another's. func pendingObservation(p inventory.PendingAssignment) conditions.Observation { - return conditions.Observation{Scope: conditions.ScopeMachine, ID: fmt.Sprintf("%s.%s.%d", p.Node, p.Module, p.ID), + return conditions.Observation{Scope: conditions.ScopeMachine, ID: fmt.Sprintf("%s.%s.%s", p.Node, p.Module, p.ID), Token: kindPendingEnded, Kind: kindPendingEnded, Machine: p.Node, Severity: conditions.Warning, Resolver: conditions.ResolverOperator, Source: "pending assignments", Summary: p.Note} @@ -671,8 +665,8 @@ func clearOrphaned(ctx context.Context, keeper *conditions.Keeper, inv *inventor if err != nil { return []string{fmt.Sprintf("the open conditions could not be read to clear orphaned ones: %v", err)} } - byID := map[int64]string{} - var ids []int64 + byID := map[string]string{} + var ids []string for _, c := range open { if c.Kind != kindPendingEnded { continue @@ -682,8 +676,8 @@ func clearOrphaned(ctx context.Context, keeper *conditions.Keeper, inv *inventor if len(parts) < 2 { continue } - var id int64 - if _, err := fmt.Sscan(parts[len(parts)-2], &id); err != nil { + id := parts[len(parts)-2] + if !strings.HasPrefix(id, "p-") { continue } byID[id] = c.Key diff --git a/cmd/mesh-controller/pending_integration_test.go b/cmd/mesh-controller/pending_integration_test.go index bbb10749..5bbf1287 100644 --- a/cmd/mesh-controller/pending_integration_test.go +++ b/cmd/mesh-controller/pending_integration_test.go @@ -637,8 +637,9 @@ func TestARaisedRowIsKeptUntilItsConditionClears(t *testing.T) { } settlePending(ctx, open, time.Now()) settlePending(ctx, open, time.Now().Add(inventory.KeptFor+time.Hour)) - if known, err := inv.PendingKnown(ctx, []int64{row.ID}); err != nil || known[row.ID] { - t.Fatalf("a cleared row outlived %s: %v", inventory.KeptFor, err) + // No key of the chain is removed (ADR 0408 §4): a cleared wait stays in the value of its key, saying it was cleared. + if known, err := inv.PendingKnown(ctx, []string{row.ID}); err != nil || !known[row.ID] || rowOf(t, open, row.ID).Cleared == nil { + t.Fatalf("a cleared row is not kept, saying so: %v", err) } // On anchor, then anchor removed: the row goes, and its condition with it at the next tick. diff --git a/cmd/mesh-controller/pending_test.go b/cmd/mesh-controller/pending_test.go index d949c02c..c5171bca 100644 --- a/cmd/mesh-controller/pending_test.go +++ b/cmd/mesh-controller/pending_test.go @@ -23,7 +23,7 @@ func conditionOpen(t *testing.T, p inventory.PendingAssignment) bool { return found } -func rowOf(t *testing.T, open *stores, id int64) inventory.PendingAssignment { +func rowOf(t *testing.T, open *stores, id string) inventory.PendingAssignment { t.Helper() rows, err := open.inventory.Pending(t.Context(), time.Unix(0, 0)) if err != nil { @@ -34,7 +34,7 @@ func rowOf(t *testing.T, open *stores, id int64) inventory.PendingAssignment { return r } } - t.Fatalf("pending assignment %d is not on record", id) + t.Fatalf("pending assignment %s is not on record", id) return inventory.PendingAssignment{} } diff --git a/internal/chain/pending_test.go b/internal/chain/pending_test.go new file mode 100644 index 00000000..c0e73f9d --- /dev/null +++ b/internal/chain/pending_test.go @@ -0,0 +1,63 @@ +package chain + +import ( + "encoding/json" + "strings" + "testing" + "time" +) + +// **A pending value is not a standing assignment** (ADR 0261, ADR 0408, issue 1094): it names its node and module, holds no +// time assigned, says `state: pending`, and every reader that skips a value with a state skips it. +func TestAPendingValueIsNotAStandingAssignment(t *testing.T) { + since := time.Date(2026, 10, 18, 3, 0, 0, 0, time.UTC) + a := Assignment{Node: "anchor", Module: "notes", State: StatePending, + Pending: &Pending{ID: "p-1", State: PendingWaiting, Build: "build-1", Since: since}} + if a.Stands() { + t.Fatal("a pending value stands") + } + raw, _ := json.Marshal(a) + for _, want := range []string{`"node":"anchor"`, `"module":"notes"`, `"state":"pending"`, `"pending":{"id":"p-1","state":"waiting"`} { + if !strings.Contains(string(raw), want) { + t.Errorf("the value lacks %s: %s", want, raw) + } + } + if strings.Contains(string(raw), `"assigned"`) { + t.Errorf("a pending value names a time assigned: %s", raw) + } + // The wait carried into the value that stands: no state, so it stands, and the record stays. + made := Assignment{Node: "anchor", Module: "notes", Assigned: since, Pending: &Pending{ID: "p-1", State: PendingApplied}} + if !made.Stands() || !a.Pending.Open() || made.Pending.Open() { + t.Fatalf("stands %v, open before %v, open after %v", made.Stands(), a.Pending.Open(), made.Pending.Open()) + } + var back Assignment + if err := json.Unmarshal(raw, &back); err != nil || back.Pending == nil || back.Pending.Build != "build-1" || !back.Pending.Since.Equal(since) { + t.Fatalf("read back %+v: %v", back, err) + } + now := since + raised := Pending{ID: "p-2", State: PendingExpired, Raised: &now, Acknowledged: &now, Cleared: &now} + body, _ := json.Marshal(raised) + if !strings.Contains(string(body), `"raised"`) || !strings.Contains(string(body), `"acknowledged"`) || !strings.Contains(string(body), `"cleared"`) { + t.Errorf("the marks of a wait: %s", body) + } +} + +func TestTheTransitionsOfAPendingAssignment(t *testing.T) { + ok := [][2]string{{PendingWaiting, PendingExpired}, {PendingWaiting, PendingWithdrawn}, {PendingApplying, PendingApplied}, + {PendingApplying, PendingRefused}} + for _, p := range ok { + if !CanSettle(p[0], p[1]) { + t.Errorf("%s may end as %s", p[0], p[1]) + } + } + for _, p := range [][2]string{{PendingWaiting, PendingApplied}, {PendingWaiting, PendingRefused}, {PendingApplying, PendingExpired}, + {PendingApplying, PendingWithdrawn}, {PendingApplied, PendingExpired}, {PendingExpired, PendingWithdrawn}, + {PendingWaiting, PendingWaiting}} { + if CanSettle(p[0], p[1]) { + t.Errorf("%s may not end as %s", p[0], p[1]) + } + } + if !CanMove(PendingWaiting, PendingApplying) || !CanMove(PendingApplying, PendingWaiting) || CanMove(PendingApplied, PendingWaiting) { + t.Error("a claim moves a wait to applying and back") + } +} diff --git a/internal/chain/records.go b/internal/chain/records.go index 709aa833..54facfc1 100644 --- a/internal/chain/records.go +++ b/internal/chain/records.go @@ -189,9 +189,62 @@ type Assignment struct { // Ended is when the assignment ended (an unassignment): absent while it stands. **No key of the chain is removed** // (ADR 0408 §4), so an unassigned module keeps its key, saying so; every reader treats a value with Ended, or with a - // State, as no assignment. State is, in a later delivery, the pending assignment waiting for its build (issue 1094). + // State, as no assignment. State is StatePending while the key holds an assignment waiting for its build, or one that + // was never made (issue 1094). Ended time.Time `json:"ended,omitzero"` State string `json:"state,omitempty"` + // Pending is the assignment's wait for its build (novox/hq ADR 0261), kept in the value of the key it waits for and, when + // it is made, carried into the value that stands: the record of what became of the wait. + Pending *Pending `json:"pending,omitempty"` +} + +// StatePending is the State of a value that holds no assignment but a pending one. +const StatePending = "pending" + +// The states of a pending assignment (novox/hq ADR 0261). +const ( + PendingWaiting = "waiting" + PendingApplying = "applying" + PendingApplied = "applied" + PendingRefused = "refused" + PendingExpired = "expired" + PendingWithdrawn = "withdrawn" +) + +// Pending is an assignment kept until its module is registered: the build it waits for, where that build is from, and the +// times of its claim and its end. Its id is p-, unique for a key, which holds one at a time. +type Pending struct { + ID string `json:"id"` + State string `json:"state"` + Build string `json:"build,omitempty"` + Repository string `json:"repository,omitempty"` + Path string `json:"path,omitempty"` + Since time.Time `json:"since"` + Note string `json:"note,omitempty"` + Settled *time.Time `json:"settled,omitempty"` + Raised *time.Time `json:"raised,omitempty"` + Acknowledged *time.Time `json:"acknowledged,omitempty"` + Cleared *time.Time `json:"cleared,omitempty"` +} + +// Open says the assignment is still to be made. +func (p Pending) Open() bool { return p.State == PendingWaiting || p.State == PendingApplying } + +// CanSettle says a pending assignment in state from may end in state to: a wait expires or is withdrawn, a claim becomes +// applied or refused, and nothing else. +func CanSettle(from, to string) bool { + switch from { + case PendingWaiting: + return to == PendingExpired || to == PendingWithdrawn + case PendingApplying: + return to == PendingApplied || to == PendingRefused + } + return false +} + +// CanMove says a pending assignment in state from may go to state to without ending: claimed, or released again. +func CanMove(from, to string) bool { + return (from == PendingWaiting && to == PendingApplying) || (from == PendingApplying && to == PendingWaiting) } // Stands says the value is an assignment that holds: it has not ended and is not waiting in a state of its own. diff --git a/internal/inventory/chain.go b/internal/inventory/chain.go index 148d22da..a56e784a 100644 --- a/internal/inventory/chain.go +++ b/internal/inventory/chain.go @@ -147,3 +147,21 @@ func (i *Inventory) BeginSwitchForTest(kind string) { i.moved.switching.Store(true) } } + +// theChain is the chain this inventory writes and reads, opened if a command has not yet: for what has no table of its own +// to be read from while it is moved (a pending assignment is a value of the assignments bucket from the start). +func (i *Inventory) theChain(ctx context.Context) (*chain.Chain, error) { + i.moved.mu.Lock() + defer i.moved.mu.Unlock() + if i.chain == nil && i.moved.open != nil { + c, err := i.moved.open(ctx) + if err != nil { + return nil, fmt.Errorf("%w: %v", ErrNoChain, err) + } + i.chain = c + } + if i.chain == nil { + return nil, ErrNoChain + } + return i.chain, nil +} diff --git a/internal/inventory/chain_assignments.go b/internal/inventory/chain_assignments.go index 958924e0..6ca7e82c 100644 --- a/internal/inventory/chain_assignments.go +++ b/internal/inventory/chain_assignments.go @@ -175,7 +175,9 @@ func (i *Inventory) chainAssign(ctx context.Context, c *chain.Chain, node Node, cur.Value.Assigned = now _, err = c.Assignments.Update(ctx, key, cur.Value, cur.Revision) case found: - // One that ended is assigned again: a fresh value over it, the ended one staying in history. + // One that ended is assigned again, and one that was pending is made: a fresh value over it, the earlier one + // staying in history. The wait is carried into the value that stands (issue 1094). + fresh.Pending = carriedPending(cur.Value, now) _, err = c.Assignments.Update(ctx, key, fresh, cur.Revision) default: _, err = c.Assignments.Create(ctx, key, fresh) @@ -384,6 +386,8 @@ func keepStep(has bool, cur, next chain.Assignment) string { return keepSkip // assigned since the bucket was read: its own write is not here yet case !has: return keepCreate + case cur.State == chain.StatePending: + return keepSkip // a wait for a build is not pulled in by a plan case !cur.Stands(): if !cur.Assigned.IsZero() || !next.Assigned.IsZero() { return keepSkip // an assignment that ended is not pulled in again by a plan @@ -394,3 +398,17 @@ func keepStep(has bool, cur, next chain.Assignment) string { } return keepField } + +// carriedPending is the wait a value that does not stand carries into the assignment made over it: a wait that was applying +// stays applying, for the one making it settles it applied; one still waiting is made by hand, and says so. Nothing for a +// value that held no wait, or whose key is not one that was pending. +func carriedPending(was chain.Assignment, now time.Time) *chain.Pending { + if was.Pending == nil || was.State != chain.StatePending { + return nil + } + p := *was.Pending + if p.State == chain.PendingWaiting { + p.State, p.Note, p.Settled = chain.PendingApplied, "assigned by hand", &now + } + return &p +} diff --git a/internal/inventory/chain_assignments_key_test.go b/internal/inventory/chain_assignments_key_test.go index 62e87cc7..839404f1 100644 --- a/internal/inventory/chain_assignments_key_test.go +++ b/internal/inventory/chain_assignments_key_test.go @@ -51,18 +51,21 @@ func TestAPassThatKeepsThePlansNeverRevivesAnEndedAssignment(t *testing.T) { ended := chain.Assignment{Node: "a", Module: "m", Assigned: at, Ended: at.Add(time.Minute)} pulledIn := chain.Assignment{Node: "a", Module: "p", RequiredBy: "m", Provisioning: plan} pulledInEnded := chain.Assignment{Node: "a", Module: "p", Ended: at} + waiting := chain.Assignment{Node: "a", Module: "p", State: chain.StatePending, + Pending: &chain.Pending{ID: "p-1", State: chain.PendingWaiting}} for name, tc := range map[string]struct { has bool cur, next chain.Assignment want string }{ - "the interleaving: wanted assigned, ended since": {true, ended, assigned, keepSkip}, - "wanted assigned, not in the bucket yet": {false, chain.Assignment{}, assigned, keepSkip}, - "a pulled-in module, not in the bucket": {false, chain.Assignment{}, pulledIn, keepCreate}, - "a pulled-in module that ended, pulled again": {true, pulledInEnded, pulledIn, keepFresh}, - "an assignment ended, a plan for it": {true, ended, chain.Assignment{Node: "a", Module: "m", RequiredBy: "x", Provisioning: plan}, keepSkip}, - "the plan is the same": {true, assigned, assigned, keepSkip}, - "the plan changed": {true, chain.Assignment{Node: "a", Module: "m", Assigned: at}, assigned, keepField}, + "the interleaving: wanted assigned, ended since": {true, ended, assigned, keepSkip}, + "wanted assigned, not in the bucket yet": {false, chain.Assignment{}, assigned, keepSkip}, + "a pulled-in module, not in the bucket": {false, chain.Assignment{}, pulledIn, keepCreate}, + "a pulled-in module that ended, pulled again": {true, pulledInEnded, pulledIn, keepFresh}, + "an assignment ended, a plan for it": {true, ended, chain.Assignment{Node: "a", Module: "m", RequiredBy: "x", Provisioning: plan}, keepSkip}, + "a wait for a build, a plan for the module pulled in": {true, waiting, pulledIn, keepSkip}, + "the plan is the same": {true, assigned, assigned, keepSkip}, + "the plan changed": {true, chain.Assignment{Node: "a", Module: "m", Assigned: at}, assigned, keepField}, } { if got := keepStep(tc.has, tc.cur, tc.next); got != tc.want { t.Errorf("%s: %s, want %s", name, got, tc.want) diff --git a/internal/inventory/chain_pending_test.go b/internal/inventory/chain_pending_test.go new file mode 100644 index 00000000..1e996403 --- /dev/null +++ b/internal/inventory/chain_pending_test.go @@ -0,0 +1,221 @@ +//go:build integration + +package inventory + +import ( + "errors" + "fmt" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/novox/mesh-controller/internal/chain" +) + +// **A pending assignment is the value of the key of the assignment it waits for** (ADR 0261, ADR 0408, issue 1094): state +// pending, the wait in its `pending` object under an id that is a string, no table, and one open wait per machine and module. +func TestAPendingAssignmentIsAKeyOfTheAssignmentsBucket(t *testing.T) { + inv, node := aNodeWithModules(t, "web") + ctx := t.Context() + p, err := inv.RecordPending(ctx, PendingAssignment{Node: node, Module: "sensors", Build: "build-1", Repository: "novox/mesh-catalog", + Path: "/modules/sensors/"}) + if err != nil { + t.Fatal(err) + } + if want := fmt.Sprintf("p-%d", p.Since.UnixNano()); p.ID != want || p.State != PendingWaiting || p.Path != "modules/sensors" { + t.Fatalf("recorded %+v, want id %s", p, want) + } + kept, found, err := inv.chain.Assignments.Get(ctx, "anchor.sensors") + if err != nil || !found || kept.Value.State != chain.StatePending || kept.Value.Stands() || !kept.Value.Assigned.IsZero() || + kept.Value.Pending == nil || kept.Value.Pending.ID != p.ID || kept.Value.Pending.Build != "build-1" { + t.Fatalf("the key holds %+v (found %v, %v)", kept.Value, found, err) + } + rows, err := inv.Pending(ctx, time.Time{}) + if err != nil || len(rows) != 1 || rows[0].ID != p.ID || rows[0].Node != "anchor" || rows[0].Build != "build-1" { + t.Fatalf("waiting: %+v %v", rows, err) + } + if _, err := inv.RecordPending(ctx, PendingAssignment{Node: node, Module: "sensors", Build: "build-2"}); !errors.Is(err, ErrAlreadyPending) { + t.Fatalf("a second open wait: %v", err) + } + if _, err := inv.Assign(ctx, node, "web"); err != nil { + t.Fatal(err) + } + if _, err := inv.RecordPending(ctx, PendingAssignment{Node: node, Module: "web", Build: "build-3"}); !errors.Is(err, ErrOnTheMachine) { + t.Fatalf("a wait for a module that is on the machine: %v", err) + } + // Written over once it ended: a settled wait is replaced by a new one under a new id. + if ok, err := inv.SettlePending(ctx, p.ID, PendingWaiting, PendingExpired, "boom"); err != nil || !ok { + t.Fatalf("settled %v %v", ok, err) + } + next, err := inv.RecordPending(ctx, PendingAssignment{Node: node, Module: "sensors", Build: "build-2"}) + if err != nil || next.ID == p.ID { + t.Fatalf("a wait after one ended: %+v %v", next, err) + } + if known, _ := inv.PendingKnown(ctx, []string{p.ID, next.ID}); known[p.ID] || !known[next.ID] { + t.Fatalf("known %v: the key holds the newer one now", known) + } + if history, _ := inv.chain.Assignments.History(ctx, "anchor.sensors"); len(history) != 2 { + t.Fatalf("the earlier wait is in the key's history: %d values", len(history)) + } + if n := rowsInTheTable(t, inv, "pending_assignment"); n != "0" { + t.Fatalf("the table took %s pending assignments", n) + } +} + +// **A pending value is skipped by what reads assignments** (issue 1094): the machine is assigned nothing by it, the module +// runs nowhere, and a module only waiting for a build is not forgotten as assigned. +func TestARecordedPendingIsSkippedByAssignedAndTheJoins(t *testing.T) { + inv, node := aNodeWithModules(t, "web") + ctx := t.Context() + if _, err := inv.RecordPending(ctx, PendingAssignment{Node: node, Module: "web", Build: "b-1"}); err != nil { + t.Fatal(err) + } + if on, _ := inv.Assigned(ctx, node); len(on) != 0 { + t.Fatalf("assigned %v", on) + } + if running, _ := inv.Running(ctx, "web"); len(running) != 0 { + t.Fatalf("running %v", running) + } + if runs, _ := inv.AssignmentRuns(ctx, node); len(runs) != 0 { + t.Fatalf("runs %v", runs) + } + entries, _ := inv.Catalogued(ctx) + for _, e := range entries { + if e.Manifest.Module == "web" && len(e.On) != 0 { + t.Fatalf("web is on %v", e.On) + } + } + if err := inv.ForgetModule(ctx, "web"); errors.Is(err, ErrStillAssigned) { + t.Fatalf("a module only waiting is assigned: %v", err) + } +} + +// **Two claims of one wait make one, and the other sees false** (issue 1094): by compare-and-set on the key's revision. +func TestTwoClaimsOfOnePendingMakeOneAndTheOtherSeesFalse(t *testing.T) { + inv, node := aNodeWithModules(t) + ctx := t.Context() + p, err := inv.RecordPending(ctx, PendingAssignment{Node: node, Module: "sensors", Build: "b-1"}) + if err != nil { + t.Fatal(err) + } + var won atomic.Int32 + var wg sync.WaitGroup + for k := 0; k < 8; k++ { + wg.Add(1) + go func() { + defer wg.Done() + if ok, err := inv.ClaimPending(ctx, p.ID); err == nil && ok { + won.Add(1) + } + }() + } + wg.Wait() + if won.Load() != 1 { + t.Fatalf("%d claims won", won.Load()) + } + if again, err := inv.ClaimPending(ctx, p.ID); err != nil || again { + t.Fatalf("a claim of a claimed wait: %v %v", again, err) + } + if rows, _ := inv.Stuck(ctx, time.Now().Add(time.Minute)); len(rows) != 1 || rows[0].State != PendingApplying || rows[0].Settled == nil { + t.Fatalf("stuck %+v", rows) + } + if err := inv.ReleaseClaim(ctx, p.ID); err != nil { + t.Fatal(err) + } + if rows, _ := inv.Pending(ctx, time.Time{}); len(rows) != 1 || rows[0].Settled != nil { + t.Fatalf("released %+v", rows) + } + // Only a wait that waits is withdrawn or expired, only a claim is applied. + if ok, _ := inv.SettlePending(ctx, p.ID, PendingWaiting, PendingApplied, "x"); ok { + t.Fatal("a wait was applied") + } + if ok, _ := inv.SettlePending(ctx, p.ID, PendingApplying, PendingRefused, "x"); ok { + t.Fatal("a wait that is not claimed was refused as a claim") + } + if ok, err := inv.RepointPending(ctx, p.ID, "b-2", "novox/other", "x/"); err != nil || !ok { + t.Fatalf("repoint %v %v", ok, err) + } + if rows, _ := inv.Pending(ctx, time.Time{}); rows[0].Build != "b-2" || rows[0].Path != "x" { + t.Fatalf("repointed %+v", rows) + } +} + +// **Assigning over a wait carries the wait into the value that stands** (issue 1094): the claim stays applying for the one +// making it to settle applied, and one still waiting is made by hand, and says so. +func TestAssignOverAPendingKeepsItsRecordAsApplied(t *testing.T) { + inv, node := aNodeWithModules(t, "web", "cache") + ctx := t.Context() + claimed, _ := inv.RecordPending(ctx, PendingAssignment{Node: node, Module: "web", Build: "b-1"}) + if ok, err := inv.ClaimPending(ctx, claimed.ID); err != nil || !ok { + t.Fatal(ok, err) + } + if added, err := inv.Assign(ctx, node, "web"); err != nil || !added { + t.Fatalf("assigned %v %v", added, err) + } + standing, _, _ := inv.chain.Assignments.Get(ctx, "anchor.web") + if !standing.Value.Stands() || standing.Value.Assigned.IsZero() || standing.Value.Pending == nil || + standing.Value.Pending.State != PendingApplying { + t.Fatalf("the value that stands: %+v", standing.Value) + } + if ok, err := inv.SettlePending(ctx, claimed.ID, PendingApplying, PendingApplied, "made"); err != nil || !ok { + t.Fatalf("settled %v %v", ok, err) + } + if on, _ := inv.Assigned(ctx, node); len(on) != 1 || on[0] != "web" { + t.Fatalf("assigned %v", on) + } + rows, _ := inv.Pending(ctx, time.Now().Add(-time.Hour)) + if len(rows) != 1 || rows[0].State != PendingApplied || rows[0].Note != "made" { + t.Fatalf("what became of the wait: %+v", rows) + } + + waiting, _ := inv.RecordPending(ctx, PendingAssignment{Node: node, Module: "cache", Build: "b-2"}) + if _, err := inv.Assign(ctx, node, "cache"); err != nil { + t.Fatal(err) + } + got, _ := inv.PendingFor(ctx, node, "cache") + if len(got) != 1 || got[0].ID != waiting.ID || got[0].State != PendingApplied || !strings.Contains(got[0].Note, "by hand") { + t.Fatalf("a wait assigned by hand: %+v", got) + } +} + +// **Marks are stamped once, and a wait of a machine the store dropped is no wait** (issue 1094). +func TestAPendingIsMarkedOnceAndGoesWithItsMachine(t *testing.T) { + inv, node := aNodeWithModules(t) + ctx := t.Context() + p, _ := inv.RecordPending(ctx, PendingAssignment{Node: node, Module: "sensors", Build: "b-1"}) + if _, err := inv.SettlePending(ctx, p.ID, PendingWaiting, PendingExpired, "boom"); err != nil { + t.Fatal(err) + } + if rows, _ := inv.ToRaise(ctx); len(rows) != 1 { + t.Fatalf("to raise %+v", rows) + } + if err := inv.MarkPending(ctx, p.ID, "raised"); err != nil { + t.Fatal(err) + } + first, _ := inv.Pending(ctx, time.Now().Add(-time.Hour)) + time.Sleep(2 * time.Millisecond) + if err := inv.MarkPending(ctx, p.ID, "raised"); err != nil { + t.Fatal(err) + } + second, _ := inv.Pending(ctx, time.Now().Add(-time.Hour)) + if !first[0].Raised.Equal(*second[0].Raised) { + t.Fatalf("the first stamp did not stand: %v then %v", first[0].Raised, second[0].Raised) + } + if err := inv.MarkPending(ctx, p.ID, "frobnicated"); err == nil { + t.Fatal("a mark nobody knows") + } + if rows, _ := inv.ToRaise(ctx); len(rows) != 0 { + t.Fatalf("to raise %+v", rows) + } + if rows, _ := inv.ToClear(ctx); len(rows) != 1 { + t.Fatalf("to clear %+v", rows) + } + if _, err := inv.RemoveNodeForTest(ctx, node); err != nil { + t.Fatal(err) + } + if rows, _ := inv.Pending(ctx, time.Unix(0, 0)); len(rows) != 0 { + t.Fatalf("a wait outlived its machine: %+v", rows) + } +} diff --git a/internal/inventory/pending.go b/internal/inventory/pending.go index 7f8258d5..ff41e0f6 100644 --- a/internal/inventory/pending.go +++ b/internal/inventory/pending.go @@ -3,7 +3,9 @@ package inventory import ( "context" "errors" + "fmt" "path" + "sort" "strings" "time" @@ -214,17 +216,19 @@ func (i *Inventory) requests(ctx context.Context) ([]RequestOutcome, error) { // The states of a pending assignment. const ( - PendingWaiting = "waiting" - PendingApplying = "applying" - PendingApplied = "applied" - PendingRefused = "refused" - PendingExpired = "expired" - PendingWithdrawn = "withdrawn" + PendingWaiting = chain.PendingWaiting + PendingApplying = chain.PendingApplying + PendingApplied = chain.PendingApplied + PendingRefused = chain.PendingRefused + PendingExpired = chain.PendingExpired + PendingWithdrawn = chain.PendingWithdrawn ) -// PendingAssignment is an assignment kept until its module is registered (ADR 0261). +// PendingAssignment is an assignment kept until its module is registered (ADR 0261): the value of its key in the +// assignments bucket while it waits (state pending), and the record carried into the value that stands once it is made. type PendingAssignment struct { - ID int64 + // ID is p-, made when it was recorded. + ID string Node string Module string // Build is the build it waits for; Repository and Path where that build is from. @@ -247,195 +251,288 @@ func (p PendingAssignment) Open() bool { // ErrAlreadyPending is a second pending assignment of one module to one machine. var ErrAlreadyPending = errors.New("that assignment is already waiting for its build") -const pendingColumns = `p.id, n.name, p.module, p.build_id, p.repository, p.source_path, p.since, p.state, p.note, - p.settled_at, p.raised_at, p.acknowledged_at, p.cleared_at` +// ErrOnTheMachine is a pending assignment of a module that is on the machine already. +var ErrOnTheMachine = errors.New("that module is on the machine already") -func scanPending(rows pgx.Rows) ([]PendingAssignment, error) { - defer rows.Close() - var out []PendingAssignment - for rows.Next() { - var p PendingAssignment - if err := rows.Scan(&p.ID, &p.Node, &p.Module, &p.Build, &p.Repository, &p.Path, &p.Since, &p.State, - &p.Note, &p.Settled, &p.Raised, &p.Acknowledged, &p.Cleared); err != nil { - return nil, err - } - out = append(out, p) - } - return out, rows.Err() +func pendingOf(v chain.Assignment) PendingAssignment { + p := v.Pending + return PendingAssignment{ID: p.ID, Node: v.Node, Module: v.Module, Build: p.Build, Repository: p.Repository, Path: p.Path, + Since: p.Since, State: p.State, Note: p.Note, Settled: p.Settled, Raised: p.Raised, Acknowledged: p.Acknowledged, + Cleared: p.Cleared} } -// RecordPending keeps an assignment until its module is registered. One is open per machine and module. +// pendingValues is every value of the bucket that holds a pending assignment, on a machine the store knows (a removed +// machine's is none, as the node's foreign key made it), with its key and revision. +func (i *Inventory) pendingValues(ctx context.Context) ([]chain.Kept[chain.Assignment], error) { + c, err := i.theChain(ctx) + if err != nil { + return nil, err + } + all, err := c.Assignments.All(ctx) + if err != nil { + return nil, fmt.Errorf("the pending assignments cannot be read from the bus: %w", err) + } + nodes, err := i.nodeNames(ctx) + if err != nil { + return nil, err + } + var out []chain.Kept[chain.Assignment] + for key, k := range all { + if k.Value.Pending != nil && nodes[k.Value.Node] { + k.Key = key + out = append(out, k) + } + } + sort.Slice(out, func(a, b int) bool { + x, y := out[a].Value.Pending, out[b].Value.Pending + if !x.Since.Equal(y.Since) { + return x.Since.Before(y.Since) + } + return x.ID < y.ID + }) + return out, nil +} + +func pendingList(kept []chain.Kept[chain.Assignment], keep func(chain.Pending) bool) []PendingAssignment { + var out []PendingAssignment + for _, k := range kept { + if keep(*k.Value.Pending) { + out = append(out, pendingOf(k.Value)) + } + } + return out +} + +// changePending changes the pending assignment of an id by compare-and-set on the key that holds it, read again when it +// moved: false when no key holds that id now, or change says no. +func (i *Inventory) changePending(ctx context.Context, id string, change func(v *chain.Assignment) bool) (bool, error) { + c, err := i.theChain(ctx) + if err != nil { + return false, err + } + for try := 0; try < casTries; try++ { + held, err := i.pendingValues(ctx) + if err != nil { + return false, err + } + var at *chain.Kept[chain.Assignment] + for k := range held { + if held[k].Value.Pending.ID == id { + at = &held[k] + } + } + if at == nil { + return false, nil + } + v := at.Value + p := *v.Pending + v.Pending = &p + if !change(&v) { + return false, nil + } + if _, err := c.Assignments.Update(ctx, at.Key, v, at.Revision); errors.Is(err, chain.ErrMoved) { + continue + } else if err != nil { + return false, err + } + return true, nil + } + return false, fmt.Errorf("the pending assignment %s kept moving under the write", id) +} + +// RecordPending keeps an assignment until its module is registered, as the value of its key (state pending). One is open per +// machine and module; a key that holds the module on the machine refuses it, and one that holds an ended assignment or a +// settled wait is written over. func (i *Inventory) RecordPending(ctx context.Context, p PendingAssignment) (PendingAssignment, error) { node, err := i.NodeByName(ctx, p.Node) if err != nil { return PendingAssignment{}, err } - err = i.store.Pool().QueryRow(ctx, - `insert into pending_assignment (node, module, build_id, repository, source_path) - values ($1, $2, $3, $4, $5) - on conflict (node, module) where state in ('waiting', 'applying') do nothing - returning id, since`, - node.ID, p.Module, p.Build, p.Repository, strings.Trim(p.Path, "/")).Scan(&p.ID, &p.Since) - if errors.Is(err, pgx.ErrNoRows) { - return PendingAssignment{}, ErrAlreadyPending - } + c, err := i.theChain(ctx) if err != nil { return PendingAssignment{}, err } - p.State = PendingWaiting - return p, nil + key := assignmentKey(node.Name, p.Module) + for try := 0; try < casTries; try++ { + cur, found, err := c.Assignments.Get(ctx, key) + if err != nil { + return PendingAssignment{}, err + } + switch { + case found && cur.Value.Stands(): + return PendingAssignment{}, fmt.Errorf("%w: %s on %s", ErrOnTheMachine, p.Module, node.Name) + case found && cur.Value.Pending != nil && cur.Value.State == chain.StatePending && cur.Value.Pending.Open(): + return PendingAssignment{}, ErrAlreadyPending + } + since := time.Now().UTC() + p.ID, p.Since, p.State = fmt.Sprintf("p-%d", since.UnixNano()), since, PendingWaiting + p.Path = strings.Trim(p.Path, "/") + value := chain.Assignment{Node: node.Name, Module: p.Module, State: chain.StatePending, Pending: &chain.Pending{ + ID: p.ID, State: PendingWaiting, Build: p.Build, Repository: p.Repository, Path: p.Path, Since: since}} + if found { + _, err = c.Assignments.Update(ctx, key, value, cur.Revision) + } else { + _, err = c.Assignments.Create(ctx, key, value) + } + if errors.Is(err, chain.ErrMoved) || errors.Is(err, chain.ErrExists) { + continue + } + if err != nil { + return PendingAssignment{}, err + } + return p, nil + } + return PendingAssignment{}, fmt.Errorf("the assignment of %s to %s kept moving under the write", p.Module, node.Name) } // RepointPending makes a waiting pending assignment wait for another build. False when it no longer waits. -func (i *Inventory) RepointPending(ctx context.Context, id int64, build, repository, sourcePath string) (bool, error) { - tag, err := i.store.Pool().Exec(ctx, - `update pending_assignment set build_id = $2, repository = $3, source_path = $4 - where id = $1 and state = 'waiting'`, id, build, repository, strings.Trim(sourcePath, "/")) - if err != nil { - return false, err - } - return tag.RowsAffected() == 1, nil +func (i *Inventory) RepointPending(ctx context.Context, id string, build, repository, sourcePath string) (bool, error) { + return i.changePending(ctx, id, func(v *chain.Assignment) bool { + if v.Pending.State != PendingWaiting { + return false + } + v.Pending.Build, v.Pending.Repository, v.Pending.Path = build, repository, strings.Trim(sourcePath, "/") + return true + }) } // Pending is every pending assignment still waiting, oldest first, when endedSince is zero. Otherwise it is // every open one and every one ended since then. func (i *Inventory) Pending(ctx context.Context, endedSince time.Time) ([]PendingAssignment, error) { - if endedSince.IsZero() { - rows, err := i.store.Pool().Query(ctx, `select `+pendingColumns+` - from pending_assignment p join node n on n.id = p.node - where p.state = 'waiting' order by p.since, p.id`) - if err != nil { - return nil, err - } - return scanPending(rows) - } - rows, err := i.store.Pool().Query(ctx, `select `+pendingColumns+` - from pending_assignment p join node n on n.id = p.node - where p.state in ('waiting', 'applying') or p.settled_at >= $1 - order by p.since, p.id`, endedSince) + held, err := i.pendingValues(ctx) if err != nil { return nil, err } - return scanPending(rows) + if endedSince.IsZero() { + return pendingList(held, func(p chain.Pending) bool { return p.State == PendingWaiting }), nil + } + return pendingList(held, func(p chain.Pending) bool { + return p.Open() || (p.Settled != nil && !p.Settled.Before(endedSince)) + }), nil } -// PendingFor is every pending assignment of one module to one machine, newest first. +// PendingFor is the pending assignment of one module to one machine: the one its key holds now, which is the newest +// (an earlier one is read from the key's history, which nothing here needs). func (i *Inventory) PendingFor(ctx context.Context, node, module string) ([]PendingAssignment, error) { - rows, err := i.store.Pool().Query(ctx, `select `+pendingColumns+` - from pending_assignment p join node n on n.id = p.node - where n.name = $1 and p.module = $2 order by p.since desc, p.id desc`, node, module) + held, err := i.pendingValues(ctx) if err != nil { return nil, err } - return scanPending(rows) + var out []PendingAssignment + for _, k := range held { + if k.Value.Node == node && k.Value.Module == module { + out = append(out, pendingOf(k.Value)) + } + } + return out, nil } // Stuck is every pending assignment left applying since before the time given: a controller that claimed // it and stopped before it said what came of it. func (i *Inventory) Stuck(ctx context.Context, before time.Time) ([]PendingAssignment, error) { - rows, err := i.store.Pool().Query(ctx, `select `+pendingColumns+` - from pending_assignment p join node n on n.id = p.node - where p.state = 'applying' and p.settled_at < $1 order by p.id`, before) + held, err := i.pendingValues(ctx) if err != nil { return nil, err } - return scanPending(rows) + return pendingList(held, func(p chain.Pending) bool { + return p.State == PendingApplying && p.Settled != nil && p.Settled.Before(before) + }), nil } // ToRaise is every pending assignment that expired or was refused and has not been raised as a condition; // ToClear every one raised and not cleared. func (i *Inventory) ToRaise(ctx context.Context) ([]PendingAssignment, error) { - rows, err := i.store.Pool().Query(ctx, `select `+pendingColumns+` - from pending_assignment p join node n on n.id = p.node - where p.state in ('expired', 'refused') and p.raised_at is null order by p.id`) + held, err := i.pendingValues(ctx) if err != nil { return nil, err } - return scanPending(rows) + return pendingList(held, func(p chain.Pending) bool { + return (p.State == PendingExpired || p.State == PendingRefused) && p.Raised == nil + }), nil } func (i *Inventory) ToClear(ctx context.Context) ([]PendingAssignment, error) { - rows, err := i.store.Pool().Query(ctx, `select `+pendingColumns+` - from pending_assignment p join node n on n.id = p.node - where p.raised_at is not null and p.cleared_at is null order by p.id`) + held, err := i.pendingValues(ctx) if err != nil { return nil, err } - return scanPending(rows) + return pendingList(held, func(p chain.Pending) bool { return p.Raised != nil && p.Cleared == nil }), nil } // MarkPending stamps a pending assignment raised, acknowledged or cleared, now. -func (i *Inventory) MarkPending(ctx context.Context, id int64, what string) error { - column := map[string]string{"raised": "raised_at", "acknowledged": "acknowledged_at", "cleared": "cleared_at"}[what] - if column == "" { +func (i *Inventory) MarkPending(ctx context.Context, id string, what string) error { + if what != "raised" && what != "acknowledged" && what != "cleared" { return errors.New("a pending assignment is marked raised, acknowledged or cleared") } - _, err := i.store.Pool().Exec(ctx, - `update pending_assignment set `+column+` = coalesce(`+column+`, now()) where id = $1`, id) + now := time.Now().UTC() + _, err := i.changePending(ctx, id, func(v *chain.Assignment) bool { + mark := map[string]**time.Time{"raised": &v.Pending.Raised, "acknowledged": &v.Pending.Acknowledged, "cleared": &v.Pending.Cleared}[what] + if *mark != nil { + return false // the first stamp stands + } + *mark = &now + return true + }) return err } // ClaimPending takes a waiting pending assignment for making, so nothing else settles or withdraws it while // it is made. False when it no longer waits: another process took it, or it was withdrawn. -func (i *Inventory) ClaimPending(ctx context.Context, id int64) (bool, error) { - tag, err := i.store.Pool().Exec(ctx, - `update pending_assignment set state = 'applying', settled_at = now() where id = $1 and state = 'waiting'`, id) - if err != nil { - return false, err - } - return tag.RowsAffected() == 1, nil +func (i *Inventory) ClaimPending(ctx context.Context, id string) (bool, error) { + now := time.Now().UTC() + return i.changePending(ctx, id, func(v *chain.Assignment) bool { + if !chain.CanMove(v.Pending.State, PendingApplying) { + return false + } + v.Pending.State, v.Pending.Settled = PendingApplying, &now + return true + }) } // ReleaseClaim puts a claimed pending assignment back to waiting. -func (i *Inventory) ReleaseClaim(ctx context.Context, id int64) error { - _, err := i.store.Pool().Exec(ctx, - `update pending_assignment set state = 'waiting', settled_at = null where id = $1 and state = 'applying'`, id) +func (i *Inventory) ReleaseClaim(ctx context.Context, id string) error { + _, err := i.changePending(ctx, id, func(v *chain.Assignment) bool { + if v.Pending.State != PendingApplying { + return false + } + v.Pending.State, v.Pending.Settled = PendingWaiting, nil + return true + }) return err } // SettlePending ends a pending assignment in the state it is read in (from): waiting for an expiry or a // withdrawal, applying for the outcome of making it. False when it was no longer in that state. -func (i *Inventory) SettlePending(ctx context.Context, id int64, from, state, note string) (bool, error) { +func (i *Inventory) SettlePending(ctx context.Context, id string, from, state, note string) (bool, error) { if state == PendingWaiting || state == PendingApplying { return false, errors.New("a pending assignment is settled to applied, refused, expired or withdrawn") } - tag, err := i.store.Pool().Exec(ctx, - `update pending_assignment set state = $3, note = $4, settled_at = now() - where id = $1 and state = $2`, id, from, state, note) - if err != nil { - return false, err - } - return tag.RowsAffected() == 1, nil + now := time.Now().UTC() + return i.changePending(ctx, id, func(v *chain.Assignment) bool { + if v.Pending.State != from || !chain.CanSettle(from, state) { + return false + } + v.Pending.State, v.Pending.Note, v.Pending.Settled = state, note, &now + return true + }) } -// ForgetEndedPending deletes ended pending assignments settled more than KeptFor ago, and says how many. One -// raised as a condition is kept until that condition is cleared: deleting it would leave the condition -// with nothing to answer it. -func (i *Inventory) ForgetEndedPending(ctx context.Context, now time.Time) (int64, error) { - tag, err := i.store.Pool().Exec(ctx, - `delete from pending_assignment where state not in ('waiting', 'applying') and settled_at < $1 - and (raised_at is null or cleared_at is not null)`, - now.Add(-KeptFor)) - if err != nil { - return 0, err - } - return tag.RowsAffected(), nil -} - -// PendingKnown says which of these pending assignments are still on record. -func (i *Inventory) PendingKnown(ctx context.Context, ids []int64) (map[int64]bool, error) { - rows, err := i.store.Pool().Query(ctx, `select id from pending_assignment where id = any($1)`, ids) +// PendingKnown says which of these pending assignments are still on record: a key holds them now. One a newer pending +// assignment of its key replaced is not. +func (i *Inventory) PendingKnown(ctx context.Context, ids []string) (map[string]bool, error) { + held, err := i.pendingValues(ctx) if err != nil { return nil, err } - defer rows.Close() - out := map[int64]bool{} - for rows.Next() { - var id int64 - if err := rows.Scan(&id); err != nil { - return nil, err - } - out[id] = true + want := map[string]bool{} + for _, id := range ids { + want[id] = true } - return out, rows.Err() + out := map[string]bool{} + for _, k := range held { + if want[k.Value.Pending.ID] { + out[k.Value.Pending.ID] = true + } + } + return out, nil } diff --git a/testdata/store-tests b/testdata/store-tests index 91c80206..8b3ec1c9 100644 --- a/testdata/store-tests +++ b/testdata/store-tests @@ -155,3 +155,8 @@ ./cmd/mesh-controller TestOneReadOfAnAssignmentNamesItsCommitBuildAndRolloutAndOneReadOfTheRolloutListsItsAssignments ./cmd/mesh-controller TestASendOfWhatAMachineRunsAlreadyChangesNoLinkOfItsAssignment ./internal/inventory TestARowChangedAfterTheFirstPassIsInTheBucketWhenTheKindIsRecordedMoved +./internal/inventory TestAPendingAssignmentIsAKeyOfTheAssignmentsBucket +./internal/inventory TestARecordedPendingIsSkippedByAssignedAndTheJoins +./internal/inventory TestTwoClaimsOfOnePendingMakeOneAndTheOtherSeesFalse +./internal/inventory TestAssignOverAPendingKeepsItsRecordAsApplied +./internal/inventory TestAPendingIsMarkedOnceAndGoesWithItsMachine