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 a2a9ccb1..266f37d7 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, pinsBucket} + engineReadsBucket, retiredBucket, copiesBucket, writtenBucket, buckets.Waits, buckets.PutWaits, 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. @@ -121,8 +121,9 @@ var keptStates = []announce.State{ {Name: copiesState, Readable: true}, {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) + {Name: buckets.PutWaitsState, Readable: true}, // the put waits said (putwaits.go, design 47 G A-211) + {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. @@ -197,6 +198,11 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any logf("the bucket %s cannot be opened (%v): no outside wait is kept until the assigner starts again", name, err) continue } + if name == buckets.PutWaits && err != nil { + // New with A-211: without it no put's wait is said until the assigner starts again. + logf("the bucket %s cannot be opened (%v): no put's wait is said until the assigner starts again", name, err) + continue + } if name == retiredBucket && err != nil { // The retired state is read only for the copy into `assignments` (migrate.go): a mesh made after the switch has // none, and nothing is copied. @@ -221,7 +227,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, pinsBucket: + buckets.Puts, buckets.Source, buckets.Merges, engineReadsBucket, retiredBucket, writtenBucket, buckets.Waits, buckets.PutWaits, buckets.Inputs, pinsBucket: return true } return slices.Contains(seatBuckets, name) || name == buckets.Grants @@ -351,6 +357,7 @@ var watched = []string{ engineReadsBucket, writtenBucket, buckets.Waits, + buckets.PutWaits, } // follow watches every bucket a pass reads and makes a pass at start, once every watch handed what stands, and after diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go index 9bf93aef..d710656a 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go @@ -63,8 +63,6 @@ type putActs struct { rec meshhandactsclient.Recorder warrantNo map[string]string said map[string]string - // waiting is each put without hold whose wait-held was said, by key (putwaits.go); nil until the first pass seeds it. - waiting map[string]putWaiting // snapshot takes the bus's snapshot on node; a variable so a test takes none. snapshot func(ctx context.Context, node, module string) (string, error) } @@ -109,8 +107,7 @@ func (p *passer) puts(ctx context.Context) { } p.placementPuts(ctx) all := p.putsByModule() - p.acts.seedPutWaits(all) - defer func() { p.putWaits(p.putsByModule()) }() + defer func() { p.putWaits(ctx, p.putsByModule()) }() if len(all) == 0 { return } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/putwaits.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/putwaits.go index 1aa45bcd..5aca048a 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/putwaits.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/putwaits.go @@ -1,12 +1,15 @@ package main import ( + "context" + "encoding/json" "fmt" "time" meshwants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" "mesh-delivery-assigner/internal/assignment" + "mesh-delivery-assigner/internal/buckets" "mesh-delivery-assigner/internal/puts" ) @@ -38,47 +41,42 @@ func putWaitBody(node, module string, p puts.Put) meshwants.Wait { Words: fmt.Sprintf("the put of %s at %s on %s waits for %s to run it", module, shortFP(p.FP), node, node)} } -// putWaiting is a put whose wait-held was said: its identity and the put as it was then. +// putWaiting is a put whose wait-held was said, as the state put-waits keeps it under the put's key (design 47 G A-204, +// A-211): its identity and the put as it was then. type putWaiting struct { - id string - put puts.Put + ID string `json:"id"` + Put puts.Put `json:"put"` } -// seedPutWaits takes, at the process's first pass, every put that waits as said already: it was said by the process before. -func (a *putActs) seedPutWaits(all map[string]map[string]puts.Put) { - a.mu.Lock() - defer a.mu.Unlock() - if a.waiting != nil { +// saidPutWaits is every put wait the state put-waits holds, by key; a value that does not decode is said and left. +func (p *passer) saidPutWaits() map[string]putWaiting { + out := map[string]putWaiting{} + for key, value := range p.w.read(buckets.PutWaits) { + var w putWaiting + if err := json.Unmarshal(value, &w); err != nil || w.ID == "" { + p.logf("%s %s holds no put wait, left: %v", buckets.PutWaits, key, err) + continue + } + out[key] = w + } + return out +} + +// putWaits says wait-ended of each put whose said wait the state put-waits keeps and that no longer waits, whether it ended +// while this process ran or while the assigner was stopped (A-204), and wait-held of each put that waits and was not said; +// one not said, or not kept, is said at the next pass. Without the state no put's wait is said. +func (p *passer) putWaits(ctx context.Context, all map[string]map[string]puts.Put) { + if p.kvs[buckets.PutWaits] == nil { return } - a.waiting = map[string]putWaiting{} - for module, byNode := range all { - for node, p := range byNode { - if key, err := puts.Key(node, module); err == nil && putWaits(p) { - a.waiting[key] = putWaiting{id: identity(key, p), put: p} - } - } - } -} - -// putWaits says wait-ended of each put that waited and no longer does, and wait-held of each put that waits and was not said; -// one not said is said at the next pass. -func (p *passer) putWaits(all map[string]map[string]puts.Put) { - a := p.acts - a.seedPutWaits(all) - a.mu.Lock() - waiting := make(map[string]putWaiting, len(a.waiting)) - for k, v := range a.waiting { - waiting[k] = v - } - a.mu.Unlock() + waiting := p.saidPutWaits() for key, was := range waiting { node, module, err := puts.ParseKey(key) if err != nil { continue } cur, there := all[module][node] - same := there && identity(key, cur) == was.id + same := there && identity(key, cur) == was.ID if same && putWaits(cur) { continue } @@ -88,15 +86,14 @@ func (p *passer) putWaits(all map[string]map[string]puts.Put) { } else if same && cur.Step == puts.Released { how = putWaitReleased } - body := putWaitBody(node, module, was.put) + body := putWaitBody(node, module, was.Put) body.How = how - if err := p.wantsSay(meshwants.EventWaitEnded, body, was.id+".put."+how); err != nil { + if err := p.wantsSay(meshwants.EventWaitEnded, body, was.ID+".put."+how); err != nil { p.logf("the end of the put %s's wait is not said, and is said at the next pass: %v", key, err) continue } - a.mu.Lock() - delete(a.waiting, key) - a.mu.Unlock() + p.remove(ctx, buckets.PutWaits, key) + delete(waiting, key) } for module, byNode := range all { for node, put := range byNode { @@ -105,19 +102,15 @@ func (p *passer) putWaits(all map[string]map[string]puts.Put) { continue } id := identity(key, put) - a.mu.Lock() - said := a.waiting[key].id == id - a.mu.Unlock() - if said { + if w, said := waiting[key]; said && w.ID == id { continue } if err := p.wantsSay(meshwants.EventWaitHeld, putWaitBody(node, module, put), id+".put"); err != nil { p.logf("the put %s's wait is not said, and is said at the next pass: %v", key, err) continue } - a.mu.Lock() - a.waiting[key] = putWaiting{id: id, put: put} - a.mu.Unlock() + kept := putWaiting{ID: id, Put: put} + p.put(ctx, buckets.PutWaits, key, func() ([]byte, error) { return json.Marshal(kept) }) } } } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/putwaits_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/putwaits_test.go index 596889ea..1106d2a7 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/putwaits_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/putwaits_test.go @@ -1,47 +1,67 @@ package main import ( + "context" "testing" "time" meshwants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" + "mesh-delivery-assigner/internal/buckets" "mesh-delivery-assigner/internal/puts" ) // Design 47 G §498 and the main file's *A put without hold that stands*: a put without hold whose want is written is said as // wait-held, step put, with its node, module, fingerprint, since and a 30-minute bound, and wait-ended when it runs, is -// released or is put again; a held put is said as no wait; a put that waits at the process's start was said before and is -// not said again; assigner_puts answers since and bound. +// released or is put again; a held put is said as no wait; assigner_puts answers since and bound. A-204, A-211: the said +// waits are kept in the state put-waits, so a put that waits at the start was said before and is not said again, and one +// that ran while the assigner was stopped gets its wait-ended at the next start. func TestAPutWithoutHoldIsSaidAsAWait(t *testing.T) { at := time.Date(2026, 11, 12, 8, 0, 0, 0, time.UTC) - p := newPasser(kvSet{}, at) - p.acts = newPutActs(&fakeBus{calls: make(chan string, 8)}, func() {}, func(string, ...any) {}) + kept := &memKV{values: map[string][]byte{}} var events []string var bodies []meshwants.Wait - p.publishWants = func(event string, body any, id string) error { - events, bodies = append(events, event), append(bodies, body.(meshwants.Wait)) - return nil + start := func() *passer { // a process's start: a fresh world that hands what the state holds + p := newPasser(kvSet{buckets.PutWaits: kept}, at) + for k, v := range kept.values { + p.w.put(buckets.PutWaits, k, v) + } + p.acts = newPutActs(&fakeBus{calls: make(chan string, 8)}, func() {}, func(string, ...any) {}) + p.publishWants = func(event string, body any, id string) error { + events, bodies = append(events, event), append(bodies, body.(meshwants.Wait)) + return nil + } + return p } + ctx := context.Background() asked := puts.Put{FP: fpTest, Why: "a try", At: at}.Said(at, puts.Asked, "asked") wanted := asked.Said(at.Add(time.Minute), puts.Wanted, "the want is written").Said(at.Add(2*time.Minute), puts.Wanted, "its check failed") held := wanted held.Hold = true - // At the start a put that waits was said before: nothing is said of it. - p.putWaits(map[string]map[string]puts.Put{"nats": {"ace": wanted}}) + p := start() + p.putWaits(ctx, map[string]map[string]puts.Put{"nats": {"ace": wanted}}) + if len(events) != 1 || events[0] != meshwants.EventWaitHeld || kept.values["ace.nats"] == nil { + t.Fatalf("a put that waits said %v, kept %q", events, kept.values) + } + events, bodies = nil, nil + // At the next start the put still waits: it was said before, nothing is said of it. + p = start() + p.putWaits(ctx, map[string]map[string]puts.Put{"nats": {"ace": wanted}}) if len(events) != 0 { t.Fatalf("a put waiting at the start was said again: %v", events) } - // It runs: wait-ended, ran. - p.putWaits(map[string]map[string]puts.Put{"nats": {"ace": wanted.Said(at.Add(3*time.Minute), puts.Done, "runs")}}) - if len(events) != 1 || events[0] != meshwants.EventWaitEnded || bodies[0].How != putWaitRan || bodies[0].Step != meshwants.WaitStepPut { - t.Fatalf("a put that ran said %v %+v", events, bodies) + // It ran while the assigner was stopped: at the next start, wait-ended, ran, and the state forgets it. + p = start() + p.putWaits(ctx, map[string]map[string]puts.Put{"nats": {"ace": wanted.Said(at.Add(3*time.Minute), puts.Done, "runs")}}) + if len(events) != 1 || events[0] != meshwants.EventWaitEnded || bodies[0].How != putWaitRan || bodies[0].Step != meshwants.WaitStepPut || + kept.values["ace.nats"] != nil { + t.Fatalf("a put that ran while stopped said %v %+v, kept %q", events, bodies, kept.values) } events, bodies = nil, nil // A new put's want is written: wait-held, since the want was written, bound 30m; a held put says nothing. again := wanted again.At = at.Add(time.Hour) - p.putWaits(map[string]map[string]puts.Put{"nats": {"ace": again}, "web": {"ace": held}}) + p.putWaits(ctx, map[string]map[string]puts.Put{"nats": {"ace": again}, "web": {"ace": held}}) if len(events) != 1 || events[0] != meshwants.EventWaitHeld { t.Fatalf("a put's written want said %v", events) } @@ -51,14 +71,14 @@ func TestAPutWithoutHoldIsSaidAsAWait(t *testing.T) { t.Errorf("the put's wait is %+v", b) } // Said once. - p.putWaits(map[string]map[string]puts.Put{"nats": {"ace": again}, "web": {"ace": held}}) + p.putWaits(ctx, map[string]map[string]puts.Put{"nats": {"ace": again}, "web": {"ace": held}}) if len(events) != 1 { t.Errorf("a put's wait was said again: %v", events) } // Put again under its key: the first wait ends superseded, the new one is held. newer := asked newer.At = at.Add(2 * time.Hour) - p.putWaits(map[string]map[string]puts.Put{"nats": {"ace": newer}}) + p.putWaits(ctx, map[string]map[string]puts.Put{"nats": {"ace": newer}}) if len(events) != 2 || events[1] != meshwants.EventWaitEnded || bodies[1].How != putWaitSuperseded || bodies[1].Fingerprint != fpTest { t.Errorf("a put asked again said %v %+v", events, bodies) } diff --git a/modules/mesh-delivery-assigner/internal/buckets/buckets.go b/modules/mesh-delivery-assigner/internal/buckets/buckets.go index 0cf5be2d..b3bb4ca0 100644 --- a/modules/mesh-delivery-assigner/internal/buckets/buckets.go +++ b/modules/mesh-delivery-assigner/internal/buckets/buckets.go @@ -25,6 +25,10 @@ const ( // Waits holds the waits on an outside cause, one key `..` (design 47 G §706). WaitsState = "waits" Waits = plc.Module + "_" + WaitsState + // PutWaits holds each put wait the assigner said, one key `.`, its value the wait's id and the put as said + // (design 47 G A-204, A-211), so a put that ends while the assigner is stopped gets its wait-ended at the next start. + PutWaitsState = "put-waits" + PutWaits = plc.Module + "_" + PutWaitsState // Inputs holds the inputs of the composition steps, `