diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept.go index c013f018..884723dd 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept.go @@ -34,6 +34,8 @@ type graphKeeper struct { source int64 ok bool said string + // pins removes a revision that ceased to be current once nothing pins it (pins.go); nil keeps every revision. + pins *pinKeeper } // keepGraph computes the revision of this pass and keeps it when the pass's inputs differ from the last computed. @@ -73,6 +75,9 @@ func (p *passer) keepGraph(ctx context.Context) { if !fresh { return } + if g.pins != nil { + g.pins.judge() // the revision before ceased to be current + } id := fmt.Sprintf("%s-%s-%d", meshwants.EventAssignmentRevision, rev.Kind, rev.Revision) if err := p.wantsSay(meshwants.EventAssignmentRevision, assigngraph.Event(rev), id); err != nil { p.logf("the assignment revision %d is kept but not said: %v", rev.Revision, err) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index baa43787..a2a9ccb1 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -80,7 +80,7 @@ var readFamilies = []string{buckets.ComposedRead} var ownBuckets = []string{buckets.Asks, buckets.Differences, buckets.Puts, buckets.Wants, plc.AssignmentsBucket, plc.ProposalsBucket, plc.CandidatesBucket, plc.PulsesBucket, assign.FindingsBucket, assign.PoliciesBucket, plc.SeatHoldersBucket, - engineReadsBucket, retiredBucket, copiesBucket, writtenBucket, buckets.Waits, buckets.Inputs} + engineReadsBucket, retiredBucket, copiesBucket, writtenBucket, buckets.Waits, buckets.Inputs, pinsBucket} // engineReadsState is the assigner's state of what each node's engine states it reads (ADR 0618 §3), one key ``, as // the verb manifest-reads last answered it or the fallback read it: what assigner_engine_reads lists and asks again. @@ -122,6 +122,7 @@ var keptStates = []announce.State{ {Name: writtenState, Readable: true}, {Name: buckets.WaitsState, Readable: true}, {Name: buckets.InputsState, Readable: true}, // read by the tool process alone: the verb input and assigner_inputs (design 47 G §354) + {Name: pinsState, Readable: true}, // the records pinning each assignment graph revision (design 47 G §579) } // kvSet is every bucket the assigner opened, by name. @@ -185,6 +186,11 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any logf("the bucket %s cannot be opened (%v): it is read by the tool process alone", name, err) continue } + if name == pinsBucket && err != nil { + // New with the pins (design 47 G §579): without it no pin is kept and no revision is removed. + logf("the bucket %s cannot be opened (%v): no pin is kept and no revision removed until it starts again", name, err) + continue + } if name == buckets.Waits && err != nil { // New with the wait on an outside cause (design 47 G §706): without it no wait is made, and a failing check is judged // as before. @@ -215,7 +221,7 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any func startsWithout(name string) bool { switch name { case plc.SeatHoldersBucket, buckets.Provisions, assign.PoliciesBucket, assign.FindingsBucket, - buckets.Puts, buckets.Source, buckets.Merges, engineReadsBucket, retiredBucket, writtenBucket, buckets.Waits, buckets.Inputs: + buckets.Puts, buckets.Source, buckets.Merges, engineReadsBucket, retiredBucket, writtenBucket, buckets.Waits, buckets.Inputs, pinsBucket: return true } return slices.Contains(seatBuckets, name) || name == buckets.Grants @@ -547,7 +553,22 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, if js != nil { p.publishWants = wantsPublisher(ctx, js, thisNode) if iv != nil && iv.h != nil { - p.graph = &graphKeeper{objects: newBusObjects(js, process.Account{Node: thisNode, Module: assignment.Module}, assigngraph.RevisionsObjects), halves: iv.h.views} + objects := newBusObjects(js, process.Account{Node: thisNode, Module: assignment.Module}, assigngraph.RevisionsObjects) + p.graph = &graphKeeper{objects: objects, halves: iv.h.views} + // The pins of the revisions (pins.go, design 47 G §579): the gate's pinned, freed, heads-replaced and pins-stated + // on the hub, registered before it starts. + if kv := kvs[pinsBucket]; kv != nil { + k := &pinKeeper{ctx: ctx, st: kvEnds{ctx, kv}, objects: objects, inUse: inUseOf(w), findings: findings, now: time.Now, logf: logger.Printf} + if err := k.made(); err != nil { + logger.Printf("the state %s is not marked made, so no revision is removed until it is: %v", pinsState, err) + } + if err := hearPins(hub, k); err != nil { + logger.Printf("a pin is not heard, so no revision is removed until the assigner starts again: %v", err) + } else { + p.graph.pins = k + k.watchUnpins(kv) + } + } } } // The node-engine seat's manifest-reads-changed, for one node (ADR 0618 §4): that node is asked again, with no timer. diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/objects.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/objects.go index 9844df61..73540150 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/objects.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/objects.go @@ -87,3 +87,15 @@ func (b *busObjects) Names(ctx context.Context) ([]string, error) { } return names, nil } + +// Delete removes the object of that name; one not held is no error (design 47 G §579: a revision nothing pins is removed). +func (b *busObjects) Delete(ctx context.Context, name string) error { + o, err := b.open(ctx) + if err != nil { + return err + } + if err := o.Delete(ctx, name); err != nil && !errors.Is(err, jetstream.ErrObjectNotFound) { + return err + } + return nil +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pins.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pins.go new file mode 100644 index 00000000..0ccafce5 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pins.go @@ -0,0 +1,524 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "slices" + "sort" + "strconv" + "strings" + "sync" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" + + mdseat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/process" + + "mesh-delivery-assigner/internal/assign" + "mesh-delivery-assigner/internal/assigngraph" + "mesh-delivery-assigner/internal/assignment" +) + +// The pins of the assignment graph revisions (design 47 G §153, §178, §216, §579; A-169): the gate says on the seat +// mesh-delivery which revisions a delivery record starts (`pinned`) and stops (`freed`) pointing at, and once at each of its +// starts `pins-stated`. The assigner hears them on its durable consumer of the events stream and keeps, in its old `state` +// section `pins` (4 MiB), one key `trunk.` per assignment graph revision whose value is the list of the records pinning +// it, added or removed by compare-and-set before the event is acknowledged; a record heard twice pins once. Two more keys: +// `made`, written once when the state is first used, and `pins-stated`, the `at` of the last pins-stated applied. A revision +// is removed, its object and its key and the entry objects no kept revision lists and no want or running build names +// (§179), when its list is empty and it is not current (the newest the assigner wrote), only while the consumer is caught up +// and once a pins-stated said after `made` was applied; judged on a freed, on a revision ceasing to be current, on being +// caught up, and at start once both hold. Never on a timer, and never by reading the gate's state. + +const ( + pinsState = "pins" + pinsBucket = assignment.Module + "_" + pinsState + pinsTool = "assigner_pins" + unpinTool = "assigner_unpin" + pinsMadeKey = "made" + pinsStatedKey = "pins-stated" + pinsCapBytes = 4 << 20 + pinMissingKind = "pin-missing" +) + +// PinsValue is one value of the state pins: the records that pin the revision, and whether the revision was no longer held +// when a record pinned it. +type PinsValue struct { + Records []string `json:"records"` + Missing bool `json:"missing,omitempty"` +} + +// pinKey is the key of an assignment graph revision, its object's own name (A-169). +func pinKey(revision int64) string { return assigngraph.TrunkObject(revision) } + +// pinRevision is the revision a key of the state names, and whether it names one. +func pinRevision(key string) (int64, bool) { + n, ok := strings.CutPrefix(key, "trunk.") + if !ok { + return 0, false + } + r, err := strconv.ParseInt(n, 10, 64) + return r, err == nil && r > 0 +} + +// pinObjects is the object bucket of the revisions as the pins remove from it. +type pinObjects interface { + assigngraph.Objects + Delete(ctx context.Context, name string) error +} + +// pinKeeper keeps the pins and removes what nothing pins. +type pinKeeper struct { + ctx context.Context + st endStore + objects pinObjects + // inUse says a want or a running build names the module's fingerprint (§179). + inUse func(module, fp string) bool + findings assign.Findings + now func() time.Time + logf func(string, ...any) + mu sync.Mutex + // caught is whether the last event applied left none of the consumer's messages pending. + caught bool +} + +// edit applies change to the value at key by compare-and-set, again while another write comes between. +func (k *pinKeeper) edit(key string, change func(v PinsValue, found bool) (PinsValue, bool)) error { + for { + raw, rev, err := k.st.get(key) + if err != nil { + return err + } + var v PinsValue + if raw != nil { + if err := json.Unmarshal(raw, &v); err != nil { + return fmt.Errorf("the pins of %s do not read: %w", key, err) + } + } + next, write := change(v, raw != nil) + if !write { + return nil + } + value, _ := json.Marshal(next) + if err := k.st.swap(key, value, rev); errors.Is(err, errSwapped) { + continue + } else { + return err + } + } +} + +// held says the revision's object is kept. +func (k *pinKeeper) held(revision int64) (bool, error) { + _, err := k.objects.Get(k.ctx, assigngraph.TrunkObject(revision)) + if errors.Is(err, assigngraph.ErrNoObject) { + return false, nil + } + return err == nil, err +} + +// pin adds record to each assignment graph revision of p; a revision no longer held is written missing, with the finding +// pin-missing naming the record. +func (k *pinKeeper) pin(p mdseat.Pins) error { + for _, r := range p.Assignment { + held, err := k.held(r) + if err != nil { + return err + } + err = k.edit(pinKey(r), func(v PinsValue, _ bool) (PinsValue, bool) { + if slices.Contains(v.Records, p.Record) && (held || v.Missing) { + return v, false + } + if !slices.Contains(v.Records, p.Record) { + v.Records = append(v.Records, p.Record) + slices.Sort(v.Records) + } + v.Missing = v.Missing || !held + return v, true + }) + if err != nil { + return err + } + if !held { + k.missing(r, p.Record) + } + } + return nil +} + +// free removes record from each assignment graph revision of p; a freed applied twice changes nothing. +func (k *pinKeeper) free(p mdseat.Pins) error { + for _, r := range p.Assignment { + err := k.edit(pinKey(r), func(v PinsValue, found bool) (PinsValue, bool) { + i := slices.Index(v.Records, p.Record) + if !found || i < 0 { + return v, false + } + v.Records = slices.Delete(v.Records, i, i+1) + return v, true + }) + if err != nil { + return err + } + } + return nil +} + +// missing writes the finding pin-missing, scope the assigner, id the revision. +func (k *pinKeeper) missing(revision int64, record string) { + if k.findings.Put == nil { + k.logf("the revision %d the record %s pins is not held, and no finding writer is wired to say so", revision, record) + return + } + o := assign.Observed{Scope: assignment.Module, ID: strconv.FormatInt(revision, 10), Kind: pinMissingKind, + Summary: fmt.Sprintf("the delivery record %s pins the assignment graph revision %d, which the assigner no longer holds", record, revision), + Headline: "A delivery record pins a revision the assigner no longer holds", + Explanation: fmt.Sprintf("The record %s pinned revision %d after it was removed; %s clears it.", record, revision, unpinTool), + Source: assignment.Module} + if err := k.findings.Put(o.Finding(k.now())); err != nil { + k.logf("the finding %s of revision %d is not written: %v", pinMissingKind, revision, err) + } +} + +// made writes the key made once, create-only; one there already stays. +func (k *pinKeeper) made() error { + raw, _, err := k.st.get(pinsMadeKey) + if err != nil || raw != nil { + return err + } + value, _ := json.Marshal(k.now().UTC()) + if err := k.st.swap(pinsMadeKey, value, 0); err != nil && !errors.Is(err, errSwapped) { + return err + } + return nil +} + +// timeAt is the time a key holds; false when it holds none. +func (k *pinKeeper) timeAt(key string) (time.Time, bool, error) { + raw, _, err := k.st.get(key) + if err != nil || raw == nil { + return time.Time{}, false, err + } + var t time.Time + if err := json.Unmarshal(raw, &t); err != nil { + return t, false, fmt.Errorf("the key %s does not read: %w", key, err) + } + return t, true, nil +} + +// stated keeps the at of a pins-stated applied. +func (k *pinKeeper) stated(at time.Time) error { + for { + raw, rev, err := k.st.get(pinsStatedKey) + if err != nil { + return err + } + var was time.Time + if raw != nil && json.Unmarshal(raw, &was) == nil && !at.After(was) { + return nil + } + value, _ := json.Marshal(at.UTC()) + if err := k.st.swap(pinsStatedKey, value, rev); errors.Is(err, errSwapped) { + continue + } else { + return err + } + } +} + +// statedSinceMade says a pins-stated said after the state was made was applied. +func (k *pinKeeper) statedSinceMade() (bool, error) { + made, ok, err := k.timeAt(pinsMadeKey) + if err != nil || !ok { + return false, err + } + at, ok, err := k.timeAt(pinsStatedKey) + if err != nil || !ok { + return false, err + } + return at.After(made), nil +} + +// hear applies one of the seat mesh-delivery's events and acknowledges it only once applied. +func (k *pinKeeper) hear(m jetstream.Msg) { + err := k.apply(m.Subject(), m.Data()) + if err != nil { + k.logf("%s: %v; it is heard again", m.Subject(), err) + _ = m.Nak() + return + } + _ = m.Ack() + caught := false + if meta, merr := m.Metadata(); merr == nil { + caught = meta.NumPending == 0 + } + k.mu.Lock() + was := k.caught + k.caught = caught + k.mu.Unlock() + if caught && (!was || m.Subject() != mdseat.SubjectPinned) { + k.judge() // on a freed, on a candidate's heads replaced, on pins-stated, and on being caught up + } +} + +// apply takes one event's body. +func (k *pinKeeper) apply(subject string, data []byte) error { + switch subject { + case mdseat.SubjectPinned, mdseat.SubjectFreed: + var p mdseat.Pins + if err := json.Unmarshal(data, &p); err != nil || p.Record == "" { + k.logf("%s holds no pins naming a record, and is not taken: %v", subject, err) + return nil + } + if subject == mdseat.SubjectPinned { + return k.pin(p) + } + return k.free(p) + case mdseat.SubjectPinsStated: + var s mdseat.PinsStatedEvent + if err := json.Unmarshal(data, &s); err != nil || s.At.IsZero() { + k.logf("%s holds no at, and is not taken: %v", subject, err) + return nil + } + return k.stated(s.At) + } + return nil // heads-replaced: a candidate's revision stops being current, judged after +} + +// judge removes every revision neither pinned nor current, when the rule allows a removal at all. +func (k *pinKeeper) judge() { + k.mu.Lock() + caught := k.caught + k.mu.Unlock() + if !caught { + return + } + ok, err := k.statedSinceMade() + if err != nil { + k.logf("the pins are not judged: %v", err) + return + } + if !ok { + return // no pins-stated applied since the state was made: a pin said before may not be heard yet + } + if err := k.remove(); err != nil { + k.logf("the revisions nothing pins are not removed: %v", err) + } +} + +// remove removes each kept revision that is not current and that no record pins, and then the entries left unlisted. +func (k *pinKeeper) remove() error { + names, err := k.objects.Names(k.ctx) + if err != nil { + return err + } + var kept []int64 + for _, n := range names { + if r, ok := pinRevision(n); ok { + kept = append(kept, r) + } + } + if len(kept) == 0 { + return nil + } + current := slices.Max(kept) + listed := map[string]bool{} + removed := 0 + for _, r := range kept { + raw, _, err := k.st.get(pinKey(r)) + if err != nil { + return err + } + var v PinsValue + if raw != nil { + if err := json.Unmarshal(raw, &v); err != nil { + return fmt.Errorf("the pins of %d do not read: %w", r, err) + } + } + if r == current || len(v.Records) > 0 { + rev, err := assigngraph.ReadRevision(k.ctx, k.objects, r) + if err != nil { + return err + } + for _, e := range rev.Entries { + listed[e.Digest] = true + } + continue + } + if err := k.objects.Delete(k.ctx, assigngraph.TrunkObject(r)); err != nil { + return fmt.Errorf("revision %d: %w", r, err) + } + if raw != nil { + if err := k.st.drop(pinKey(r)); err != nil { + return fmt.Errorf("the pins of %d: %w", r, err) + } + } + removed++ + } + if removed == 0 { + return nil + } + for _, n := range names { + digest, ok := strings.CutPrefix(n, "entry.") + if !ok || listed[digest] { + continue + } + e, err := assigngraph.ReadEntry(k.ctx, k.objects, digest) + if err != nil { + return err + } + if k.inUse != nil && k.inUse(e.Module, e.Fp) { + continue + } + if err := k.objects.Delete(k.ctx, n); err != nil { + return fmt.Errorf("the entry %s: %w", digest, err) + } + } + k.logf("%d assignment graph revisions no record pins and not current are removed", removed) + return nil +} + +// hearPins registers the seat mesh-delivery's four events on the assigner's hub. +func hearPins(hub *process.EventHub, k *pinKeeper) error { + for _, s := range []string{mdseat.SubjectPinned, mdseat.SubjectFreed, mdseat.SubjectHeadsReplaced, mdseat.SubjectPinsStated} { + if err := hub.Handle(s, k.hear); err != nil { + return fmt.Errorf("%s: %w", s, err) + } + } + return nil +} + +// pinsTools are assigner_pins and assigner_unpin over the runtime's states. +func pinsTools(st store, ask func(string, any) (json.RawMessage, error)) []stdio.Tool { + return []stdio.Tool{{Name: pinsTool, Description: "the assignment graph revisions the delivery records pin (design 47 G §579): per revision " + + "the records that pin it and whether it was missing when pinned, the newest revision kept (current, never removed), when the state " + + "was made, the last pins-stated applied, and the state's fill against its cap of 4 MiB.", + Run: func(map[string]any) (any, error) { return readPins(st) }}, + {Name: unpinTool, Description: "Remove one record from a revision's pins, for a freed that was lost (design 47 G §579); written in the " + + "hand-act log first. The removal rule then applies.", + Input: schema(map[string]string{"revision": "the assignment graph revision's number", "record": "the delivery record's id", + "why": "why it is removed by hand"}, []string{"revision", "record", "why"}), + Run: func(a map[string]any) (any, error) { + record, _ := a["record"].(string) + why, _ := a["why"].(string) + return unpin(st, ask, fmt.Sprint(a["revision"]), record, why, stdio.CallerOf(a)) + }}} +} + +// readPins is what assigner_pins answers. +func readPins(st store) (any, error) { + ks, err := st.keys(pinsState) + if err != nil { + return nil, err + } + sort.Strings(ks) + revisions := map[string]PinsValue{} + out := map[string]any{"revisions": revisions} + var newest int64 + bytes := 0 + for _, key := range ks { + raw, _, found, err := st.get(pinsState, key) + if err != nil || !found { + continue + } + bytes += len(key) + len(raw) + if r, ok := pinRevision(key); ok { + var v PinsValue + if json.Unmarshal(raw, &v) == nil { + revisions[key] = v + } + newest = max(newest, r) + continue + } + out[key] = json.RawMessage(raw) + } + out["fill"] = meshwants.Fill{Store: pinsState, Bytes: int64(bytes), Cap: pinsCapBytes} + if newest > 0 { + out["newest-pinned"] = newest + } + return out, nil +} + +// unpin removes one record from one revision's list, the hand act recorded first. +func unpin(st store, ask func(string, any) (json.RawMessage, error), revision, record, why, by string) (any, error) { + r, err := strconv.ParseInt(strings.TrimPrefix(revision, "trunk."), 10, 64) + switch { + case err != nil || r <= 0: + return nil, fmt.Errorf("revision %q is no assignment graph revision's number", revision) + case record == "" || why == "": + return nil, errors.New("name the record and say why") + case by == "" || by == stdio.Unproven: + return nil, errors.New("an unpin names who asks it, and the runtime proved no caller") + } + key := pinKey(r) + raw, _, found, err := st.get(pinsState, key) + if err != nil { + return nil, err + } + var v PinsValue + if found { + if err := json.Unmarshal(raw, &v); err != nil { + return nil, err + } + } + i := slices.Index(v.Records, record) + if i < 0 { + return nil, fmt.Errorf("the record %s does not pin revision %d: %s lists its pins", record, r, pinsTool) + } + taken, err := meshhandactsclient.Recorder{Ask: ask}.Record(meshhandactsclient.Request{What: unpinTool + " " + key + " " + record, + Why: why + " (" + by + ")", Cause: "unpin"}) + if err != nil { + return nil, fmt.Errorf("the hand act is not recorded, so nothing is removed: %w", err) + } + v.Records = slices.Delete(v.Records, i, i+1) + value, _ := json.Marshal(v) + if err := st.put(pinsState, key, value); err != nil { + return nil, err + } + return map[string]any{"revision": r, "unpinned": record, "records": v.Records, "hand-act": taken.Log}, nil +} + +// watchUnpins judges again when a revision's pins change in the state itself, as assigner_unpin writes them from the tool +// process: the assigner's own state, heard as it changes, never on a timer. +func (k *pinKeeper) watchUnpins(kv jetstream.KeyValue) { + w, err := kv.WatchAll(k.ctx, jetstream.UpdatesOnly()) + if err != nil { + k.logf("the state %s is not watched, so a hand unpin is judged at the next freed or revision: %v", pinsState, err) + return + } + go func() { + defer func() { _ = w.Stop() }() + for e := range w.Updates() { + if e == nil { + continue + } + if _, ok := pinRevision(e.Key()); ok && e.Operation() == jetstream.KeyValuePut { + k.judge() + } + } + }() +} + +// inUseOf says a want or a running build names the module's fingerprint, as the world holds them now. +func inUseOf(w *world) func(module, fp string) bool { + return func(module, fp string) bool { + for _, want := range w.wantsOf()[module] { + if want.Fp == fp { + return true + } + } + for _, mods := range w.runningNow() { + if r, ok := mods[module]; ok && r.Fingerprint == fp { + return true + } + } + return false + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pins_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pins_test.go new file mode 100644 index 00000000..de19ca35 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pins_test.go @@ -0,0 +1,107 @@ +package main + +import ( + "context" + "encoding/json" + "testing" + "time" + + mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + mdseat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" + + "mesh-delivery-assigner/internal/assign" + "mesh-delivery-assigner/internal/assigngraph" +) + +func (m memObjects) Delete(_ context.Context, name string) error { delete(m, name); return nil } + +// The pins of the assignment graph revisions (design 47 G §153, §178, §579; A-169): a revision two records pin stays until +// both free it; a record pinning twice pins once; the current revision is never removed; nothing is removed before a +// pins-stated said after the state was made, nor while the consumer is behind; a revision no longer held is pinned missing, +// with the finding pin-missing; an entry another kept revision lists, or a want or a running build names, stays. +func TestPinsKeepARevisionWhileARecordPinsIt(t *testing.T) { + objects := memObjects{} + entry := func(module, fp string) string { + e := meshwants.AssignmentEntry{Node: "n1", Module: module, Fp: fp} + d := assigngraph.EntryDigest(e) + raw, _ := json.Marshal(e) + objects[assigngraph.EntryObject(d)] = raw + return d + } + shared, old, running := entry("a", "fp-a"), entry("b", "fp-b1"), entry("c", "fp-c1") + revision := func(n int64, digests ...string) { + rev := meshwants.AssignmentRevision{Revision: n} + for _, d := range digests { + rev.Entries = append(rev.Entries, meshwants.NodeEntryRef{Node: "n1", Digest: d}) + } + raw, _ := json.Marshal(rev) + objects[assigngraph.TrunkObject(n)] = raw + } + revision(1, shared, old, running) + revision(2, shared, old) + revision(3, shared) + var found []mcseat.Finding + clock := time.Date(2026, 11, 12, 10, 0, 0, 0, time.UTC) + k := &pinKeeper{ctx: context.Background(), st: newMemEnds(), objects: objects, now: func() time.Time { return clock }, + logf: t.Logf, + inUse: func(module, fp string) bool { return module == "c" && fp == "fp-c1" }, + findings: assign.Findings{Put: func(f mcseat.Finding) error { found = append(found, f); return nil }}} + if err := k.made(); err != nil { + t.Fatal(err) + } + say := func(subject string, body any) { + t.Helper() + raw, _ := json.Marshal(body) + if err := k.apply(subject, raw); err != nil { + t.Fatal(err) + } + k.judge() + } + pins := func(record string, revs ...int64) mdseat.Pins { return mdseat.Pins{Record: record, Assignment: revs} } + has := func(n int64) bool { _, ok := objects[assigngraph.TrunkObject(n)]; return ok } + + k.caught = true + say(mdseat.SubjectPinned, pins("r1", 1)) + say(mdseat.SubjectPinned, pins("r1", 1)) + say(mdseat.SubjectPinned, pins("r2", 1)) + say(mdseat.SubjectFreed, pins("r9", 2)) // nothing pins 2, but no pins-stated since made: nothing removed + if !has(2) { + t.Fatal("revision 2 was removed before a pins-stated said after the state was made") + } + k.caught = false + say(mdseat.SubjectPinsStated, mdseat.PinsStatedEvent{At: clock.Add(time.Minute)}) + if !has(2) { + t.Fatal("revision 2 was removed while the consumer was behind") + } + k.caught = true + say(mdseat.SubjectFreed, pins("r1", 1)) + if has(2) { + t.Error("revision 2, neither pinned nor current, is kept") + } + if !has(1) || !has(3) { + t.Error("revision 1 (pinned by r2) or 3 (current) was removed") + } + raw, _, _ := k.st.get(pinKey(1)) + if string(raw) != `{"records":["r2"]}` { + t.Errorf("the pins of 1 = %s, want r2 alone (r1 pinned twice pins once, and freed once)", raw) + } + say(mdseat.SubjectFreed, pins("r2", 1)) + if has(1) { + t.Error("revision 1 is kept once both records freed it") + } + if _, ok := objects[assigngraph.EntryObject(old)]; ok { + t.Error("the entry no kept revision lists and nothing names is kept") + } + if _, ok := objects[assigngraph.EntryObject(running)]; !ok { + t.Error("the entry of a running build is removed") + } + if _, ok := objects[assigngraph.EntryObject(shared)]; !ok { + t.Error("the entry the current revision lists is removed") + } + say(mdseat.SubjectPinned, pins("r3", 2)) + raw, _, _ = k.st.get(pinKey(2)) + if string(raw) != `{"records":["r3"],"missing":true}` || len(found) != 1 || found[0].Kind != pinMissingKind || found[0].ID != "2" { + t.Errorf("a pin of a revision no longer held = %s, findings %+v", raw, found) + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go index 7664c4fb..879a6039 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go @@ -417,6 +417,7 @@ func serveTools() error { // What each node's engine reads, as the assigner uses it (ADR 0618 §3): its own state engine-reads, asked again through it. tools = append(tools, enginereads.Tool(engineReadsTool, "the assigner", enginereads.KeptState(stdio.State(engineReadsState)), false, time.Now)) tools = append(tools, writtenTools(runtimeStore{})...) + tools = append(tools, pinsTools(runtimeStore{}, stdio.Ask)...) tools = append(tools, askAgainTools(runtimeStore{}, stdio.Ask, time.Now)...) return stdio.Serve("", tools) } diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index 94ad01ef..49fab370 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -67,7 +67,7 @@ func TestManifest(t *testing.T) { } if want := map[string]int{"asks": 5, "differences": 64, "puts": 20, "wants": 20, "assignments": 64, "placement": 64, "copies": 1, "proposals": 5, "candidates": 5, "pulses": 0, "findings": 5, "policies": 20, "seat-holders": 5, views.MarksState: 1, views.StatusState: 1, - "engine-reads": 1, "written": 5, "waits": 5, "inputs": 1}; !reflect.DeepEqual(states, want) { + "engine-reads": 1, "written": 5, "waits": 5, "inputs": 1, "pins": 5}; !reflect.DeepEqual(states, want) { t.Errorf("state = %v, want %v", states, want) } if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 { @@ -102,7 +102,7 @@ func TestManifest(t *testing.T) { // which lists and asks again what each node's engine reads. if !slices.Equal(m.Tools, []string{"assigner_put", "assigner_release", "assigner_puts", "assigner_waits", "assigner_inputs", "assignments", "seat-holders", "wants", "assigner_read_assignments", "assigner_read_seat_holders", "assigner_read_wants", "assigner_read_findings", "assigner_read_asks", "assigner_read_differences", "assigner_read_copies", - "assigner_first_views", "assigner_first_view_again", "assigner_engine_reads", "assigner_seat_list", "assigner_seat_list_again", "assigner_ask_again", "assigner_written"}) { + "assigner_first_views", "assigner_first_view_again", "assigner_engine_reads", "assigner_seat_list", "assigner_seat_list_again", "assigner_ask_again", "assigner_written", "assigner_pins", "assigner_unpin"}) { t.Errorf("the assigner serves tools %v, want its three put tools, three first views, six read tools, assigner_engine_reads, the seat list's two (ADR 0636 §7) and assigner_ask_again (ADR 0668 §5)", m.Tools) } if m.RunsAs != m.Module || m.SecretsOwner != m.RunsAs { @@ -130,7 +130,7 @@ func TestManifest(t *testing.T) { } // ADR 0618 §4: and it hears each node's engine say its reads changed; ADR 0636 §4 (issue 1966): and the seat list's two events; // issue 1966 note 151: and the teller's ask-ended, its subject the one the assigner hears. - if !slices.Equal(m.Consumes, []string{"mesh-conditions.findings-asked", "mesh-conditions.condition-raised", "mesh-conditions.condition-changed", "mesh-conditions.condition-cleared", "mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-catalog.version-released", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", builds.BuilderModule + "." + builds.BuildEndedEvent, buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed", "node-engine.running"}) { + if !slices.Equal(m.Consumes, []string{"mesh-conditions.findings-asked", "mesh-conditions.condition-raised", "mesh-conditions.condition-changed", "mesh-conditions.condition-cleared", "mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-catalog.version-released", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", builds.BuilderModule + "." + builds.BuildEndedEvent, buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed", "node-engine.running", "mesh-delivery.pinned", "mesh-delivery.freed", "mesh-delivery.heads-replaced", "mesh-delivery.pins-stated"}) { t.Errorf("consumes = %v, want the catalog half's revisions (design 47 G §195), the teller's ask-ended, the engines' reads changed and what they run", m.Consumes) } if want := []string{"mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index fe290015..163efe1d 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -58,7 +58,11 @@ "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed", - "node-engine.running" + "node-engine.running", + "mesh-delivery.pinned", + "mesh-delivery.freed", + "mesh-delivery.heads-replaced", + "mesh-delivery.pins-stated" ], "invokes": [ "mesh-delivery-indexer.manifests", @@ -116,7 +120,9 @@ "assigner_seat_list", "assigner_seat_list_again", "assigner_ask_again", - "assigner_written" + "assigner_written", + "assigner_pins", + "assigner_unpin" ], "state": [ { "name": "assignments", "history": 64 }, @@ -137,7 +143,8 @@ { "name": "engine-reads", "history": 1 }, { "name": "written", "history": 5 }, { "name": "waits", "history": 5 }, - { "name": "inputs", "history": 1 } + { "name": "inputs", "history": 1 }, + { "name": "pins", "history": 5, "max-mib": 4 } ], "config": { "outside-wait-minutes": {