diff --git a/internal/chain/backend.go b/internal/chain/backend.go index a099ad3c..b99f3673 100644 --- a/internal/chain/backend.go +++ b/internal/chain/backend.go @@ -35,10 +35,6 @@ type Backend interface { Watch(ctx context.Context) (<-chan Entry, error) // History is a key's earlier values and its current one, oldest first, at most the bucket keeps. History(ctx context.Context, key string) ([]Entry, error) - // Remove ends a key at the revision it was read at, with a marker: **its earlier values stay**, read back by History, and - // nothing is purged. Used by an unassignment and by nothing else, in the delivery that moves the assignments; the - // delivery after it keeps an unassigned key with an end mark and removes this (issue 1096). - Remove(ctx context.Context, key string, revision uint64) error } var ( @@ -101,10 +97,10 @@ func (m *Memory) now() time.Time { return time.Now() } -// last is a key's newest entry, false when it holds none or was removed. +// last is a key's newest entry, false when it holds none. func (m *Memory) last(key string) (Entry, bool) { h := m.values[key] - if len(h) == 0 || h[len(h)-1].Deleted { + if len(h) == 0 { return Entry{}, false } return h[len(h)-1], true @@ -123,26 +119,6 @@ func (m *Memory) Get(_ context.Context, key string) (Entry, bool, error) { return clone(e), true, nil } -func (m *Memory) Remove(_ context.Context, key string, revision uint64) error { - m.mu.Lock() - defer m.mu.Unlock() - if m.fail != nil { - return m.fail - } - e, ok := m.last(key) - if !ok || e.Revision != revision { - return ErrMoved - } - m.revision++ - m.writes++ - h := append(m.values[key], Entry{Key: key, Deleted: true, Revision: m.revision, At: m.now()}) - if len(h) > m.keep() { - h = h[len(h)-m.keep():] - } - m.values[key] = h - return nil -} - func (m *Memory) Create(_ context.Context, key string, value []byte) (uint64, error) { m.mu.Lock() defer m.mu.Unlock() @@ -288,9 +264,7 @@ func (m *Memory) History(_ context.Context, key string) ([]Entry, error) { } out := make([]Entry, 0, len(m.values[key])) for _, e := range m.values[key] { - if !e.Deleted { - out = append(out, clone(e)) - } + out = append(out, clone(e)) } return out, nil } diff --git a/internal/chain/bus.go b/internal/chain/bus.go index 954813cf..c94f6373 100644 --- a/internal/chain/bus.go +++ b/internal/chain/bus.go @@ -41,7 +41,7 @@ func OnTheBus(ctx context.Context, conn *nats.Conn, holds Holds) (*Chain, error) return New(builds, rollouts, assignments, holds), nil } -// busBucket is a Backend on a key-value bucket. It calls no delete and no purge: the type has none to call. +// busBucket is a Backend on a key-value bucket. It calls no delete and no purge: the type has none to call (ADR 0408 §4). type busBucket struct{ kv jetstream.KeyValue } func entryOf(e jetstream.KeyValueEntry) Entry { @@ -76,16 +76,6 @@ func (b busBucket) Update(ctx context.Context, key string, value []byte, revisio return rev, err } -func (b busBucket) Remove(ctx context.Context, key string, revision uint64) error { - // A delete marker, never a purge: the key's earlier values stay in the bucket's history. - err := b.kv.Delete(ctx, key, jetstream.LastRevision(revision)) - var apiErr *jetstream.APIError - if errors.As(err, &apiErr) && apiErr.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequence { - return ErrMoved - } - return err -} - func (b busBucket) Keys(ctx context.Context) ([]string, error) { keys, err := b.kv.Keys(ctx) if errors.Is(err, jetstream.ErrNoKeysFound) { @@ -137,7 +127,7 @@ func (b busBucket) History(ctx context.Context, key string) ([]Entry, error) { out := make([]Entry, 0, len(all)) for _, e := range all { if e.Operation() != jetstream.KeyValuePut { - continue + continue // a marker an earlier mirror left; nothing of the chain makes one } out = append(out, entryOf(e)) } diff --git a/internal/chain/chain.go b/internal/chain/chain.go index f39a9b00..bb3f6862 100644 --- a/internal/chain/chain.go +++ b/internal/chain/chain.go @@ -182,20 +182,8 @@ type Chain struct { Rollouts Bucket[Rollout] Assignments Bucket[Assignment] - rollouts Backend - assignments Backend - holds Holds -} - -// RemoveAssignment ends an assignment's key with a marker, at the revision it was read at, only by a controller that holds -// the lease. **The one removal the chain has**, for an unassignment in the delivery that moves the assignments: the -// key's earlier values stay (history), and nothing is purged. The delivery after it keeps an unassigned key with an end -// mark in its place and removes this (issue 1096). Builds and rollouts have none. -func (c *Chain) RemoveAssignment(ctx context.Context, key string, revision uint64) error { - if err := c.Assignments.guard(ctx, key, nil); err != nil { - return err - } - return c.assignments.Remove(ctx, key, revision) + rollouts Backend + holds Holds } // RolloutsOf is the rollouts bucket typed as the controller's plan, which carries more than the chain's Rollout names: @@ -209,7 +197,6 @@ func New(builds, rollouts, assignments Backend, holds Holds) *Chain { Rollouts: Bucket[Rollout]{rollouts, holds}, Assignments: Bucket[Assignment]{assignments, holds}, rollouts: rollouts, - assignments: assignments, holds: holds, } } diff --git a/internal/chain/chain_test.go b/internal/chain/chain_test.go index de0a26ab..42c216a0 100644 --- a/internal/chain/chain_test.go +++ b/internal/chain/chain_test.go @@ -335,51 +335,57 @@ func TestAnEmptyBucketIsReadAsEmpty(t *testing.T) { } } -// **The one removal the chain has ends an assignment's key with a marker and keeps its earlier values** (issue 1093, -// ADR 0408 §5): the key reads as absent, can be made again, builds and rollouts have no removal, and a controller without -// the lease removes nothing. -func TestAnAssignmentsKeyEndsWithAMarkerAndKeepsItsHistory(t *testing.T) { +// **An assignment that has ended still stands in the bucket as a value that says so, and nothing in the chain can remove a +// key** (ADR 0408 §4, issue 1093): the value with `ended` is no assignment to a reader, and the earlier value stays in +// history. +func TestAnEndedAssignmentIsNoAssignmentAndStaysInTheBucket(t *testing.T) { ctx := context.Background() - var away error - c := New(NewMemory(), NewMemory(), NewMemory(), func(context.Context) error { return away }) - rev, err := c.Assignments.Create(ctx, "anchor.notes", Assignment{Node: "anchor", Module: "notes", Commit: "c1"}) + c := InMemory(nil) + rev, err := c.Assignments.Create(ctx, "anchor.notes", Assignment{Node: "anchor", Module: "notes", Assigned: at(1), Commit: "c1"}) if err != nil { t.Fatal(err) } - away = errors.New("another controller holds the lease") - if err := c.RemoveAssignment(ctx, "anchor.notes", rev); !errors.Is(err, ErrNotHeld) { - t.Fatalf("a removal without the lease answered %v", err) + standing, _, _ := c.Assignments.Get(ctx, "anchor.notes") + if !standing.Value.Stands() { + t.Fatal("a value that has not ended does not stand") } - away = nil - if err := c.RemoveAssignment(ctx, "anchor.notes", rev+1); !errors.Is(err, ErrMoved) { - t.Fatalf("a removal at a revision that moved answered %v", err) - } - if err := c.RemoveAssignment(ctx, "anchor.notes", rev); err != nil { + ended := standing.Value + ended.Ended = at(2) + if _, err := c.Assignments.Update(ctx, "anchor.notes", ended, rev); err != nil { t.Fatal(err) } - if _, ok, _ := c.Assignments.Get(ctx, "anchor.notes"); ok { - t.Fatal("a removed key reads as present") + got, found, _ := c.Assignments.Get(ctx, "anchor.notes") + if !found || got.Value.Stands() || got.Value.Commit != "c1" || got.Value.Ended.IsZero() { + t.Fatalf("the ended value is %+v (found %v)", got.Value, found) } - if keys, _ := c.Assignments.Keys(ctx); len(keys) != 0 { - t.Fatalf("keys %v", keys) - } - if all, _ := c.Assignments.All(ctx); len(all) != 0 { - t.Fatalf("all %v", all) - } - if history, _ := c.Assignments.History(ctx, "anchor.notes"); len(history) != 1 || history[0].Value.Commit != "c1" { + if history, _ := c.Assignments.History(ctx, "anchor.notes"); len(history) != 2 || !history[0].Value.Stands() { t.Fatalf("the earlier values are %+v", history) } - if _, err := c.Assignments.Create(ctx, "anchor.notes", Assignment{Node: "anchor", Module: "notes"}); err != nil { - t.Fatalf("a removed key could not be made again: %v", err) + if (Assignment{Node: "a", Module: "m", Assigned: at(1), State: "pending"}).Stands() { + t.Fatal("a pending value stands") } - if _, ok := any(c.Builds).(interface { - Remove(context.Context, string, uint64) error - }); ok { - t.Fatal("builds can be removed") + raw, _ := json.Marshal(ended) + if !strings.Contains(string(raw), `"ended":"2026-10-18T02:00:00Z"`) { + t.Fatalf("the ended value is %s", raw) } - if _, ok := any(c.Rollouts).(interface { + // And a value that stands says nothing of either, so a reader of today reads the bytes it did. + raw, _ = json.Marshal(Assignment{Node: "a", Module: "m", Assigned: at(1)}) + if strings.Contains(string(raw), "ended") || strings.Contains(string(raw), "state") { + t.Fatalf("a standing value says %s", raw) + } + // No bucket, and no chain, has a removal. + type remover interface { Remove(context.Context, string, uint64) error - }); ok { - t.Fatal("rollouts can be removed") + } + for name, x := range map[string]any{"builds": c.Builds, "rollouts": c.Rollouts, "assignments": c.Assignments, "chain": c, + "a backend": NewMemory()} { + if _, has := x.(remover); has { + t.Errorf("%s can remove a key", name) + } + if _, has := x.(interface { + RemoveAssignment(context.Context, string, uint64) error + }); has { + t.Errorf("%s can remove an assignment", name) + } } } diff --git a/internal/chain/records.go b/internal/chain/records.go index c259e241..709aa833 100644 --- a/internal/chain/records.go +++ b/internal/chain/records.go @@ -186,7 +186,16 @@ type Assignment struct { Commit string `json:"commit,omitempty"` Rollout string `json:"rollout,omitempty"` Since time.Time `json:"since,omitzero"` + + // 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). + Ended time.Time `json:"ended,omitzero"` + State string `json:"state,omitempty"` } +// Stands says the value is an assignment that holds: it has not ended and is not waiting in a state of its own. +func (a Assignment) Stands() bool { return a.Ended.IsZero() && a.State == "" } + // Key is an assignment's key. func (a Assignment) Key() string { return a.Node + "." + a.Module } diff --git a/internal/inventory/chain_assignments.go b/internal/inventory/chain_assignments.go index 4ebf8fbe..67f9191b 100644 --- a/internal/inventory/chain_assignments.go +++ b/internal/inventory/chain_assignments.go @@ -16,9 +16,9 @@ import ( // Assignments on the bus (novox/hq ADR 0408, issue 1093): once the kind `assignments` is moved (chain.go) an assignment is // a key `.` of the assignments bucket, written by compare-and-set by the controller that holds the lease, -// and every question the joins of the store answered is answered from the bucket. **Live values only** in this delivery: an -// unassignment ends the key with a marker, as it removed the row, and no `ended` or `pending` mark is written; the -// delivery after it keeps an unassigned key (issue 1096). +// and every question the joins of the store answered is answered from the bucket. **No key is removed** (ADR 0408 §4): an +// unassignment writes the value again with `ended` set, and a value with `ended` or a `state` is no assignment to any reader +// here (chain.Assignment.Stands). No `pending` state is written yet (issue 1094). // KindAssignments is the kind of the chain that is each module assigned to a node, with the build it runs there. const KindAssignments = "assignments" @@ -32,8 +32,9 @@ type assignRow struct { } // assignments is every assignment, by node then module: from the bucket when the kind is moved, else from the store. -// A key whose value names no time assigned is a module only pulled in by another (issue 765): no assignment. One on a -// machine the store does not know is no assignment either, as the store took a removed machine's. +// A key whose value names no time assigned is a module only pulled in by another (issue 765), and one that has ended or +// waits in a state is gone: no assignment. One on a machine the store does not know is none either, as the store took a +// removed machine's. func (i *Inventory) assignments(ctx context.Context) ([]assignRow, error) { c, on, err := i.ChainReads(ctx, KindAssignments) if err != nil { @@ -66,7 +67,7 @@ func (i *Inventory) assignments(ctx context.Context) ([]assignRow, error) { } var out []assignRow for _, k := range all { - if k.Value.Assigned.IsZero() || !nodes[k.Value.Node] { + if k.Value.Assigned.IsZero() || !k.Value.Stands() || !nodes[k.Value.Node] { continue } out = append(out, assignRow{k.Value.Node, k.Value.Module, k.Value.Assigned}) @@ -163,16 +164,21 @@ func (i *Inventory) chainAssign(ctx context.Context, c *chain.Chain, node Node, if err != nil { return false, err } - if found && !cur.Value.Assigned.IsZero() { + if found && !cur.Value.Assigned.IsZero() && cur.Value.Stands() { return false, nil // one assignment of a module per machine: a repeat is absorbed (novox/hq ADR 0115) } now := time.Now().UTC() - if found { - // A module only pulled in by another becomes an assignment. + fresh := chain.Assignment{Node: node.Name, Module: module, Assigned: now} + switch { + case found && cur.Value.Stands(): + // A module only pulled in by another becomes an assignment, keeping what it holds. cur.Value.Assigned = now _, err = c.Assignments.Update(ctx, key, cur.Value, cur.Revision) - } else { - _, err = c.Assignments.Create(ctx, key, chain.Assignment{Node: node.Name, Module: module, Assigned: now}) + case found: + // One that ended is assigned again: a fresh value over it, the ended one staying in history. + _, err = c.Assignments.Update(ctx, key, fresh, cur.Revision) + default: + _, err = c.Assignments.Create(ctx, key, fresh) } if errors.Is(err, chain.ErrMoved) || errors.Is(err, chain.ErrExists) { continue @@ -188,7 +194,7 @@ func (i *Inventory) chainAssign(ctx context.Context, c *chain.Chain, node Node, return false, fmt.Errorf("the assignment of %s to %s kept moving under the write", module, node.Name) } -// chainUnassign takes a module off a machine: the key ends with a marker, its earlier values stay. +// chainUnassign takes a module off a machine: the value is written again with `ended`, the key stays. func (i *Inventory) chainUnassign(ctx context.Context, c *chain.Chain, node Node, module string) error { key := assignmentKey(node.Name, module) for try := 0; try < casTries; try++ { @@ -196,18 +202,16 @@ func (i *Inventory) chainUnassign(ctx context.Context, c *chain.Chain, node Node if err != nil { return err } - if !found || cur.Value.Assigned.IsZero() { + if !found || cur.Value.Assigned.IsZero() || !cur.Value.Stands() { return fmt.Errorf("%s is not assigned to %s", module, node.Name) } - if err := c.RemoveAssignment(ctx, key, cur.Revision); errors.Is(err, chain.ErrMoved) { + cur.Value.Ended = time.Now().UTC() + if _, err := c.Assignments.Update(ctx, key, cur.Value, cur.Revision); errors.Is(err, chain.ErrMoved) { continue } else if err != nil { return fmt.Errorf("%s is not taken off %s on the bus: %w", module, node.Name, err) } - if err := i.raiseGeneration(ctx); err != nil { - return err - } - return nil + return i.raiseGeneration(ctx) } return fmt.Errorf("the assignment of %s to %s kept moving under the write", module, node.Name) } @@ -281,13 +285,19 @@ func (i *Inventory) KeepAssignmentPlans(ctx context.Context) (written, removed i return written, removed, err } cur, has := have[key] - if has && sameJSON(cur.Value.Provisioning, next.Provisioning) && cur.Value.RequiredBy == next.RequiredBy { + standing := has && cur.Value.Stands() + if standing && sameJSON(cur.Value.Provisioning, next.Provisioning) && cur.Value.RequiredBy == next.RequiredBy { continue } - if !has { + switch { + case !has: next.Assigned = time.Time{} _, err = c.Assignments.Create(ctx, key, next) - } else { + case !standing: + // A value that ended, pulled in again: a fresh one over it, the ended one staying in history. + next.Assigned = time.Time{} + _, err = c.Assignments.Update(ctx, key, next, cur.Revision) + default: cur.Value.RequiredBy, cur.Value.Provisioning = next.RequiredBy, next.Provisioning _, err = c.Assignments.Update(ctx, key, cur.Value, cur.Revision) } @@ -300,11 +310,12 @@ func (i *Inventory) KeepAssignmentPlans(ctx context.Context) (written, removed i written++ } for key, cur := range have { - if _, wanted := want[key]; wanted || !cur.Value.Assigned.IsZero() { + if _, wanted := want[key]; wanted || !cur.Value.Assigned.IsZero() || !cur.Value.Stands() { continue } - // A value of a module no longer pulled in, which no assignment names. - if err := c.RemoveAssignment(ctx, key, cur.Revision); errors.Is(err, chain.ErrMoved) { + // A value of a module no longer pulled in, which no assignment names: it ends, and stays. + cur.Value.Ended = time.Now().UTC() + if _, err := c.Assignments.Update(ctx, key, cur.Value, cur.Revision); errors.Is(err, chain.ErrMoved) { continue } else if err != nil { return written, removed, fmt.Errorf("the plan of %s is not ended: %w", key, err) @@ -331,7 +342,8 @@ type KeptValue struct { Revision uint64 } -// KeptAssignments is every key of the assignments bucket with its value, and false when the assignments are not on the bus. +// KeptAssignments is every key of the assignments bucket that holds a value that stands, and false when the assignments are +// not on the bus. // What the serving controller compares from one look to the next to say a key changed (`assignment-changed`). func (i *Inventory) KeptAssignments(ctx context.Context) (map[string]KeptValue, bool, error) { c, on, err := i.ChainReads(ctx, KindAssignments) @@ -344,6 +356,9 @@ func (i *Inventory) KeptAssignments(ctx context.Context) (map[string]KeptValue, } out := make(map[string]KeptValue, len(all)) for key, k := range all { + if !k.Value.Stands() { + continue // an ended value is no assignment: for what is said of it, the key has gone + } body, err := json.Marshal(k.Value) if err != nil { return nil, true, err diff --git a/internal/inventory/chain_move_test.go b/internal/inventory/chain_move_test.go index aa8d757a..56ddf33a 100644 --- a/internal/inventory/chain_move_test.go +++ b/internal/inventory/chain_move_test.go @@ -527,8 +527,8 @@ func TestTheMoveCarriesThePlansAndTheirMergesAndTheyReadTheSame(t *testing.T) { } } -// **An assignment is a key of the assignments bucket; an unassignment ends the key and keeps its values; a repeat is -// absorbed** (ADR 0408, issue 1093): the readers of the store — what a machine runs, where a module runs, whether a module +// **An assignment is a key of the assignments bucket; an unassignment writes `ended` and keeps the key and its values; a +// repeat is absorbed** (ADR 0408, issue 1093): the readers of the store — what a machine runs, where a module runs, whether a module // may be forgotten, what is behind — read the bucket, the generation rises with each change, and a seat held by the // assignment goes with it. func TestAnAssignmentIsAKeyOfTheBucketAndAnUnassignmentEndsIt(t *testing.T) { @@ -597,12 +597,21 @@ func TestAnAssignmentIsAKeyOfTheBucketAndAnUnassignmentEndsIt(t *testing.T) { if generationNow(t, inv) <= before { t.Fatal("an unassignment left the generation where it was") } - if _, found, _ := inv.chain.Assignments.Get(ctx, "anchor.postgres"); found { - t.Fatal("an unassigned key reads as present") + // No key is removed (ADR 0408 §4): the value says it ended, and the one before it is in the key's history. + gone, found, err := inv.chain.Assignments.Get(ctx, "anchor.postgres") + if err != nil || !found || gone.Value.Ended.IsZero() || gone.Value.Stands() || gone.Value.Assigned.IsZero() { + t.Fatalf("the unassigned key: %+v %v %v", gone.Value, found, err) } - if history, _ := inv.chain.Assignments.History(ctx, "anchor.postgres"); len(history) != 1 || history[0].Value.Module != "postgres" { + if history, _ := inv.chain.Assignments.History(ctx, "anchor.postgres"); len(history) != 2 || !history[0].Value.Stands() || + history[1].Value.Stands() { t.Fatalf("the earlier values of the key: %+v", history) } + if on, _ := inv.Assigned(ctx, node); len(on) != 0 { + t.Fatalf("an ended assignment is still assigned: %v", on) + } + if running, _ := inv.Running(ctx, "postgres"); len(running) != 0 { + t.Fatalf("an ended assignment is running: %v", running) + } if held, err := inv.Holdings(ctx); err != nil || len(held) != 0 { t.Fatalf("the seat the assignment held is held by %+v: %v", held, err) } @@ -612,13 +621,26 @@ func TestAnAssignmentIsAKeyOfTheBucketAndAnUnassignmentEndsIt(t *testing.T) { if err := inv.ForgetModule(ctx, "postgres"); err != nil { t.Fatalf("a module nothing is assigned could not be forgotten: %v", err) } - // Made again after it ended. + // Made again after it ended: a fresh value over the ended one, which stays in history. if added, err := inv.Assign(ctx, node, "web"); err != nil || !added { t.Fatal(added, err) } if added, err := inv.Assign(ctx, node, "web"); err != nil || added { t.Fatal(added, err) } + if err := inv.Unassign(ctx, node, "web"); err != nil { + t.Fatal(err) + } + if added, err := inv.Assign(ctx, node, "web"); err != nil || !added { + t.Fatalf("assigned again after it ended: %v %v", added, err) + } + again, _, _ := inv.chain.Assignments.Get(ctx, "anchor.web") + if !again.Value.Stands() || !again.Value.Ended.IsZero() { + t.Fatalf("the assignment made again is %+v", again.Value) + } + if history, _ := inv.chain.Assignments.History(ctx, "anchor.web"); len(history) != 3 || history[1].Value.Stands() { + t.Fatalf("the values of the key: %+v", history) + } } // **The provisioner's plans are kept in the assignments' values, one field at a time** (issue 1093, design 57 P6): the @@ -678,8 +700,8 @@ func TestThePlansAreKeptInTheAssignmentsValuesAndNothingElseOfThemMoves(t *testi if _, removed, err := inv.KeepAssignmentPlans(ctx); err != nil || removed != 1 { t.Fatalf("ended %d: %v", removed, err) } - if _, found, _ := inv.chain.Assignments.Get(ctx, "anchor.notify"); found { - t.Fatal("a module no longer pulled in has a value") + if gone, found, _ := inv.chain.Assignments.Get(ctx, "anchor.notify"); !found || gone.Value.Stands() || gone.Value.Ended.IsZero() { + t.Fatalf("a module no longer pulled in has the value %+v, which should say it ended", gone.Value) } if on, _ := inv.Assigned(ctx, node); !reflect.DeepEqual(on, []string{"dunst", "web"}) { t.Fatalf("assigned: %v", on)