AS13: the assigner keeps the pins of its assignment graph revisions
A delivery record's pinned and freed reach the assigner on its durable consumer, kept per revision in its old state pins by compare-and-set before the event is acknowledged; a revision neither pinned nor current is removed only while the consumer is caught up and after a pins-stated said since the state was made, never on a timer and never by reading the gate's state. assigner_pins reads it, assigner_unpin frees a lost freed by hand, recorded in the hand-act log first. Builds: design 47 G §153, §178, §179, §216, §579; A-169 Left of AS13: §181's filling, full and room. Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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 `<node>`, 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.
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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.<n>` 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
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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": {
|
||||
|
||||
Reference in New Issue
Block a user