diff --git a/cmd/mesh-controller/chain_integration_test.go b/cmd/mesh-controller/chain_integration_test.go index c6a98e1c..565ca164 100644 --- a/cmd/mesh-controller/chain_integration_test.go +++ b/cmd/mesh-controller/chain_integration_test.go @@ -207,3 +207,42 @@ func runOf(on []chain.Assignment, module string) chain.Assignment { a, _ := assignmentOf(on, module) return a } + +// **An assignment waiting for its build is one key of the assignments bucket** (ADR 0261, ADR 0408, issue 1094): entered +// through the `assign` verb for a module whose build is in flight, one read of `.` says state pending with the +// build it waits for, the build's outcome through the build seat's path makes the key stand with the wait carried as applied, +// and the table of pending assignments is read by nobody. No key is written by hand. +func TestAnAssignmentWaitingForItsBuildIsOneKeyOfTheAssignmentsBucket(t *testing.T) { + open := aCatalogueMesh(t) + ctx := t.Context() + inv := open.inventory + asksWithPaths(t) + mergeAdding(t, open, "modules/sensors", "3da80a4b00aa") + if _, err := assign(ctx, open, "laptop", "sensors"); err != nil { + t.Fatalf("an assignment while its build runs was refused: %v", err) + } + waiting, found, err := inv.AssignmentValue(ctx, "laptop", "sensors") + if err != nil || !found || waiting.State != chain.StatePending || waiting.Stands() || !waiting.Assigned.IsZero() || + waiting.Pending == nil || waiting.Pending.State != chain.PendingWaiting || waiting.Pending.Build != "b-modules/sensors" || + !strings.HasPrefix(waiting.Pending.ID, "p-") { + t.Fatalf("one read of the key says %+v (found %v, %v)", waiting, found, err) + } + if got := assignedTo(t, open, "laptop"); slices.Contains(got, "sensors") { + t.Fatalf("a module only waiting is assigned: %v", got) + } + + if err := (builds{inv: inv, open: open}).Built(ctx, outcome("b-modules/sensors", "modules/sensors", "3da80a4b00aa", "sensors", "")); err != nil { + t.Fatal(err) + } + made, found, err := inv.AssignmentValue(ctx, "laptop", "sensors") + if err != nil || !found || !made.Stands() || made.State != "" || made.Assigned.IsZero() || made.Pending == nil || + made.Pending.State != chain.PendingApplied || made.Pending.ID != waiting.Pending.ID { + t.Fatalf("one read of the key after the build says %+v (found %v, %v)", made, found, err) + } + if !slices.Contains(assignedTo(t, open, "laptop"), "sensors") { + t.Fatal("the build registered sensors and it was not assigned") + } + if rows := inv.QueryForTest(t, `select count(*) from pending_assignment`); rows[0][0] != "0" { + t.Fatalf("the table of pending assignments took %s rows", rows[0][0]) + } +} diff --git a/cmd/mesh-controller/pending.go b/cmd/mesh-controller/pending.go index 977617bb..a5ec7359 100644 --- a/cmd/mesh-controller/pending.go +++ b/cmd/mesh-controller/pending.go @@ -695,7 +695,7 @@ func clearOrphaned(ctx context.Context, keeper *conditions.Keeper, inv *inventor if known[id] { continue } - if _, err := keeper.Clear(ctx, key, "its pending assignment is no longer on record: its machine was removed"); err != nil { + if _, err := keeper.Clear(ctx, key, "its pending assignment is no longer on record: its machine was removed, or the module was assigned and the wait taken back"); err != nil { said = append(said, fmt.Sprintf("the condition %s could not be cleared: %v", key, err)) } } diff --git a/internal/chain/records.go b/internal/chain/records.go index 54facfc1..ceb23533 100644 --- a/internal/chain/records.go +++ b/internal/chain/records.go @@ -225,6 +225,10 @@ type Pending struct { Raised *time.Time `json:"raised,omitempty"` Acknowledged *time.Time `json:"acknowledged,omitempty"` Cleared *time.Time `json:"cleared,omitempty"` + // Earlier are the waits of this key that ended unmade and were replaced by this one before their condition was answered + // (taken back or cleared): each stays, with its own marks, so a replaced wait never leaves the operator without the + // condition it raised or was to raise. Dropped once answered. + Earlier []Pending `json:"earlier,omitempty"` } // Open says the assignment is still to be made. diff --git a/internal/inventory/chain_assignments.go b/internal/inventory/chain_assignments.go index 6ca7e82c..3a4d058f 100644 --- a/internal/inventory/chain_assignments.go +++ b/internal/inventory/chain_assignments.go @@ -412,3 +412,13 @@ func carriedPending(was chain.Assignment, now time.Time) *chain.Pending { } return &p } + +// AssignmentValue is one key of the assignments bucket as it is held, standing or not: one read of an assignment. +func (i *Inventory) AssignmentValue(ctx context.Context, node, module string) (chain.Assignment, bool, error) { + c, err := i.theChain(ctx) + if err != nil { + return chain.Assignment{}, false, err + } + k, found, err := c.Assignments.Get(ctx, assignmentKey(node, module)) + return k.Value, found, err +} diff --git a/internal/inventory/chain_pending_test.go b/internal/inventory/chain_pending_test.go index 1e996403..2f62ee2c 100644 --- a/internal/inventory/chain_pending_test.go +++ b/internal/inventory/chain_pending_test.go @@ -219,3 +219,55 @@ func TestAPendingIsMarkedOnceAndGoesWithItsMachine(t *testing.T) { t.Fatalf("a wait outlived its machine: %+v", rows) } } + +// **A wait replaced by a newer one never leaves the operator without the condition it raised, or was to raise** (issue +// 1108): the key carries the replaced wait while its condition is unanswered, still known, still to clear, and a newer +// wait takes only the unanswered ones forward. +func TestAReplacedWaitNeverLeavesTheOperatorWithoutItsCondition(t *testing.T) { + inv, node := aNodeWithModules(t) + ctx := t.Context() + record := func(build string) PendingAssignment { + p, err := inv.RecordPending(ctx, PendingAssignment{Node: node, Module: "sensors", Build: build}) + if err != nil { + t.Fatal(err) + } + return p + } + end := func(p PendingAssignment) { + if ok, err := inv.SettlePending(ctx, p.ID, PendingWaiting, PendingExpired, "boom "+p.Build); err != nil || !ok { + t.Fatal(ok, err) + } + } + first := record("b-1") + end(first) + if err := inv.MarkPending(ctx, first.ID, "raised"); err != nil { + t.Fatal(err) + } + second := record("b-2") + if known, _ := inv.PendingKnown(ctx, []string{first.ID, second.ID}); !known[first.ID] || !known[second.ID] { + t.Fatalf("known %v: the replaced wait's condition is unanswered", known) + } + if clear, _ := inv.ToClear(ctx); len(clear) != 1 || clear[0].ID != first.ID { + t.Fatalf("to clear %+v: the first's condition is still to clear", clear) + } + rows, _ := inv.PendingFor(ctx, node, "sensors") + if len(rows) != 2 || rows[0].ID != second.ID || rows[1].ID != first.ID { + t.Fatalf("the key says %+v, newest first", rows) + } + // The second ends unmade before the first was answered: both are to be said, the first's condition not cleared by it. + end(second) + if raise, _ := inv.ToRaise(ctx); len(raise) != 1 || raise[0].ID != second.ID { + t.Fatalf("to raise %+v", raise) + } + // A third replaces the second, which was not raised yet: it is carried, with the first (still unanswered). + if err := inv.MarkPending(ctx, first.ID, "acknowledged"); err != nil { + t.Fatal(err) + } + third := record("b-3") + if raise, _ := inv.ToRaise(ctx); len(raise) != 1 || raise[0].ID != second.ID { + t.Fatalf("the second was never raised and is to be: %+v", raise) + } + if known, _ := inv.PendingKnown(ctx, []string{first.ID, second.ID, third.ID}); known[first.ID] || !known[second.ID] || !known[third.ID] { + t.Fatalf("known %v: the first was answered (taken back) and is let go, the second is not", known) + } +} diff --git a/internal/inventory/pending.go b/internal/inventory/pending.go index ff41e0f6..4d163eef 100644 --- a/internal/inventory/pending.go +++ b/internal/inventory/pending.go @@ -254,8 +254,10 @@ var ErrAlreadyPending = errors.New("that assignment is already waiting for its b // 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 pendingOf(v chain.Assignment) PendingAssignment { - p := v.Pending +func pendingOf(v chain.Assignment) PendingAssignment { return pendingRecord(v, *v.Pending) } + +// pendingRecord is one wait of a key, the one it holds or one it replaced. +func pendingRecord(v chain.Assignment, p chain.Pending) PendingAssignment { 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} @@ -296,16 +298,39 @@ func (i *Inventory) pendingValues(ctx context.Context) ([]chain.Kept[chain.Assig 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)) + for _, p := range pendingWaits(k.Value) { + if keep(p) { + out = append(out, pendingRecord(k.Value, p)) + } } } + sort.SliceStable(out, func(a, b int) bool { + if !out[a].Since.Equal(out[b].Since) { + return out[a].Since.Before(out[b].Since) + } + return out[a].ID < out[b].ID + }) 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) { +// pendingWaits is the wait a value holds and the earlier ones it carries, each without the others. +func pendingWaits(v chain.Assignment) []chain.Pending { + if v.Pending == nil { + return nil + } + out := make([]chain.Pending, 0, 1+len(v.Pending.Earlier)) + for _, e := range v.Pending.Earlier { + e.Earlier = nil + out = append(out, e) + } + held := *v.Pending + held.Earlier = nil + return append(out, held) +} + +// changePending changes the pending assignment of an id — the wait a key holds or one it carries from before — 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(p *chain.Pending) bool) (bool, error) { c, err := i.theChain(ctx) if err != nil { return false, err @@ -316,21 +341,34 @@ func (i *Inventory) changePending(ctx context.Context, id string, change func(v return false, err } var at *chain.Kept[chain.Assignment] + var target *chain.Pending for k := range held { - if held[k].Value.Pending.ID == id { - at = &held[k] + v := held[k].Value + p := *v.Pending + p.Earlier = append([]chain.Pending(nil), p.Earlier...) + v.Pending = &p + switch { + case p.ID == id: + at, target = &held[k], v.Pending + default: + for e := range v.Pending.Earlier { + if v.Pending.Earlier[e].ID == id { + at, target = &held[k], &v.Pending.Earlier[e] + } + } + } + if at != nil { + held[k].Value = v + break } } if at == nil { return false, nil } - v := at.Value - p := *v.Pending - v.Pending = &p - if !change(&v) { + if !change(target) { return false, nil } - if _, err := c.Assignments.Update(ctx, at.Key, v, at.Revision); errors.Is(err, chain.ErrMoved) { + if _, err := c.Assignments.Update(ctx, at.Key, at.Value, at.Revision); errors.Is(err, chain.ErrMoved) { continue } else if err != nil { return false, err @@ -368,7 +406,8 @@ func (i *Inventory) RecordPending(ctx context.Context, p PendingAssignment) (Pen 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}} + ID: p.ID, State: PendingWaiting, Build: p.Build, Repository: p.Repository, Path: p.Path, Since: since, + Earlier: unansweredWaits(cur.Value, found)}} if found { _, err = c.Assignments.Update(ctx, key, value, cur.Revision) } else { @@ -387,11 +426,11 @@ func (i *Inventory) RecordPending(ctx context.Context, p PendingAssignment) (Pen // RepointPending makes a waiting pending assignment wait for another build. False when it no longer waits. 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 i.changePending(ctx, id, func(p *chain.Pending) bool { + if p.State != PendingWaiting { return false } - v.Pending.Build, v.Pending.Repository, v.Pending.Path = build, repository, strings.Trim(sourcePath, "/") + p.Build, p.Repository, p.Path = build, repository, strings.Trim(sourcePath, "/") return true }) } @@ -411,8 +450,8 @@ func (i *Inventory) Pending(ctx context.Context, endedSince time.Time) ([]Pendin }), nil } -// 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). +// PendingFor is every pending assignment of one module to one machine the key still has to say, newest first: the wait it +// holds now and the unanswered ones it replaced. func (i *Inventory) PendingFor(ctx context.Context, node, module string) ([]PendingAssignment, error) { held, err := i.pendingValues(ctx) if err != nil { @@ -421,9 +460,13 @@ func (i *Inventory) PendingFor(ctx context.Context, node, module string) ([]Pend var out []PendingAssignment for _, k := range held { if k.Value.Node == node && k.Value.Module == module { - out = append(out, pendingOf(k.Value)) + for _, w := range pendingWaits(k.Value) { + out = append(out, pendingRecord(k.Value, w)) + } } } + // Newest first. + sort.SliceStable(out, func(a, b int) bool { return out[a].Since.After(out[b].Since) }) return out, nil } @@ -465,8 +508,8 @@ func (i *Inventory) MarkPending(ctx context.Context, id string, what string) err return errors.New("a pending assignment is marked raised, acknowledged or cleared") } 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] + _, err := i.changePending(ctx, id, func(p *chain.Pending) bool { + mark := map[string]**time.Time{"raised": &p.Raised, "acknowledged": &p.Acknowledged, "cleared": &p.Cleared}[what] if *mark != nil { return false // the first stamp stands } @@ -480,22 +523,22 @@ func (i *Inventory) MarkPending(ctx context.Context, id string, what string) err // 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 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 i.changePending(ctx, id, func(p *chain.Pending) bool { + if !chain.CanMove(p.State, PendingApplying) { return false } - v.Pending.State, v.Pending.Settled = PendingApplying, &now + p.State, p.Settled = PendingApplying, &now return true }) } // ReleaseClaim puts a claimed pending assignment back to waiting. 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 { + _, err := i.changePending(ctx, id, func(p *chain.Pending) bool { + if p.State != PendingApplying { return false } - v.Pending.State, v.Pending.Settled = PendingWaiting, nil + p.State, p.Settled = PendingWaiting, nil return true }) return err @@ -508,17 +551,17 @@ func (i *Inventory) SettlePending(ctx context.Context, id string, from, state, n return false, errors.New("a pending assignment is settled to applied, refused, expired or withdrawn") } now := time.Now().UTC() - return i.changePending(ctx, id, func(v *chain.Assignment) bool { - if v.Pending.State != from || !chain.CanSettle(from, state) { + return i.changePending(ctx, id, func(p *chain.Pending) bool { + if p.State != from || !chain.CanSettle(from, state) { return false } - v.Pending.State, v.Pending.Note, v.Pending.Settled = state, note, &now + p.State, p.Note, p.Settled = state, note, &now return true }) } -// 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. +// PendingKnown says which of these pending assignments are still on record: a key holds them now, or carries them as +// unanswered earlier waits. func (i *Inventory) PendingKnown(ctx context.Context, ids []string) (map[string]bool, error) { held, err := i.pendingValues(ctx) if err != nil { @@ -530,9 +573,26 @@ func (i *Inventory) PendingKnown(ctx context.Context, ids []string) (map[string] } out := map[string]bool{} for _, k := range held { - if want[k.Value.Pending.ID] { - out[k.Value.Pending.ID] = true + for _, w := range pendingWaits(k.Value) { + if want[w.ID] { + out[w.ID] = true + } } } return out, nil } + +// unansweredWaits are the waits a key holds that ended unmade and whose condition nobody has answered yet — taken back or +// cleared — which a new wait replacing them carries, so the operator is not left without it (issue 1108). +func unansweredWaits(v chain.Assignment, found bool) []chain.Pending { + if !found || v.State != chain.StatePending { + return nil + } + var out []chain.Pending + for _, w := range pendingWaits(v) { + if (w.State == PendingExpired || w.State == PendingRefused) && w.Cleared == nil && w.Acknowledged == nil { + out = append(out, w) + } + } + return out +} diff --git a/testdata/store-tests b/testdata/store-tests index 8b3ec1c9..1c1c3e58 100644 --- a/testdata/store-tests +++ b/testdata/store-tests @@ -160,3 +160,5 @@ ./internal/inventory TestTwoClaimsOfOnePendingMakeOneAndTheOtherSeesFalse ./internal/inventory TestAssignOverAPendingKeepsItsRecordAsApplied ./internal/inventory TestAPendingIsMarkedOnceAndGoesWithItsMachine +./internal/inventory TestAReplacedWaitNeverLeavesTheOperatorWithoutItsCondition +./cmd/mesh-controller TestAnAssignmentWaitingForItsBuildIsOneKeyOfTheAssignmentsBucket