From 00c888546ac44c7ecde1c9556d54b2a7a9ec927d Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 05:51:26 +0100 Subject: [PATCH] =?UTF-8?q?=C2=A7498:=20the=20assigner=20says=20a=20put=20?= =?UTF-8?q?without=20hold=20as=20a=20wait,=20and=20assigner=5Fputs=20answe?= =?UTF-8?q?rs=20since=20and=20bound?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A put without hold whose want is written is said as wait-held on the seat mesh-wants (step put, node, module, fingerprint, since, bound 30 minutes) and wait-ended when it runs, is released or is put again, so the watchman times it; a held put is no wait. Builds: design 47 G §498; main file *A put without hold that stands* Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../cmd/mesh-delivery-assigner/puts.go | 4 + .../cmd/mesh-delivery-assigner/putwaits.go | 123 ++++++++++++++++++ .../mesh-delivery-assigner/putwaits_test.go | 72 ++++++++++ .../cmd/mesh-delivery-assigner/tools.go | 25 +++- .../internal/puts/puts.go | 10 ++ 5 files changed, 230 insertions(+), 4 deletions(-) create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/putwaits.go create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/putwaits_test.go 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 dab41719..9bf93aef 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go @@ -63,6 +63,8 @@ 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) } @@ -107,6 +109,8 @@ func (p *passer) puts(ctx context.Context) { } p.placementPuts(ctx) all := p.putsByModule() + p.acts.seedPutWaits(all) + defer func() { p.putWaits(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 new file mode 100644 index 00000000..1aa45bcd --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/putwaits.go @@ -0,0 +1,123 @@ +package main + +import ( + "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/puts" +) + +// A put without hold that stands is said as a wait (design 47 G §498, main file *A put without hold that stands*): when its +// want is written the assigner says `wait-held` on the seat mesh-wants, step put, with the node, the module, the fingerprint +// and since when, bound 30 minutes; and `wait-ended` when it ends: ran when its node runs it, released by assigner_release, +// superseded by a put asked again under its key. The watchman times it; a held put waits on nobody and is said as none. + +// putWaitBound is the bound of a put without hold's wait. +const putWaitBound = 30 * time.Minute + +// Wait-ended's how of a put's wait. +const ( + putWaitRan = "ran" + putWaitReleased = "released" + putWaitSuperseded = "superseded" +) + +// putWaits says whether p is a put that waits: one without hold whose want is written and not yet run. +func putWaits(p puts.Put) bool { + return !p.Hold && p.Step == puts.Wanted +} + +// putWaitBody is the wait of the put p of module on node. +func putWaitBody(node, module string, p puts.Put) meshwants.Wait { + since := p.StandsSince().UTC() + return meshwants.Wait{Step: meshwants.WaitStepPut, Owner: assignment.Module, Node: node, Module: module, Fingerprint: p.FP, + Since: since, ClockFrom: &since, Bound: putWaitBound.String(), + 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. +type putWaiting struct { + id string + put puts.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 { + 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() + 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 + if same && putWaits(cur) { + continue + } + how := putWaitSuperseded + if same && cur.Step == puts.Done { + how = putWaitRan + } else if same && cur.Step == puts.Released { + how = putWaitReleased + } + body := putWaitBody(node, module, was.put) + body.How = how + 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() + } + for module, byNode := range all { + for node, put := range byNode { + key, err := puts.Key(node, module) + if err != nil || !putWaits(put) { + continue + } + id := identity(key, put) + a.mu.Lock() + said := a.waiting[key].id == id + a.mu.Unlock() + if said { + 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() + } + } +} 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 new file mode 100644 index 00000000..596889ea --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/putwaits_test.go @@ -0,0 +1,72 @@ +package main + +import ( + "testing" + "time" + + meshwants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" + + "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. +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) {}) + 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 + } + 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}}) + 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) + } + 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}}) + if len(events) != 1 || events[0] != meshwants.EventWaitHeld { + t.Fatalf("a put's written want said %v", events) + } + b := bodies[0] + if b.Node != "ace" || b.Module != "nats" || b.Fingerprint != fpTest || !b.Since.Equal(at.Add(time.Minute)) || b.ClockFrom == nil || + !b.ClockFrom.Equal(b.Since) || b.Bound != "30m0s" || b.Owner != "mesh-delivery-assigner" { + 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}}) + 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}}) + 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) + } + // The answer of assigner_puts. + if a := answerOf("ace", "nats", wanted); !a.Since.Equal(at.Add(time.Minute)) || a.Bound != "30m0s" { + t.Errorf("a put that waits answers since %v, bound %q", a.Since, a.Bound) + } + if a := answerOf("ace", "web", held); !a.Since.Equal(at.Add(time.Minute)) || a.Bound != "" { + t.Errorf("a held put answers since %v, bound %q", a.Since, a.Bound) + } +} 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 879a6039..a362a264 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go @@ -137,6 +137,19 @@ type putAnswer struct { // Older says a put that has not ended stands at an older version than the newest build, and which that is: it pins the // node's want until it ends or is released. Older string `json:"older-than-newest,omitempty"` + // Since is when the put came to stand at its step, held or not; Bound is the bound of a put without hold whose want is + // written, the wait the watchman times and seeds from here (design 47 G §498, main file *A put without hold that stands*). + Since time.Time `json:"since"` + Bound string `json:"bound,omitempty"` +} + +// answerOf is what assigner_put and assigner_puts answer of the put p of module on node. +func answerOf(node, module string, p puts.Put) putAnswer { + a := putAnswer{Node: node, Module: module, Put: p, Since: p.StandsSince()} + if putWaits(p) { + a.Bound = putWaitBound.String() + } + return a } // doPut is the tool assigner_put. @@ -168,7 +181,9 @@ func (s *putTools) doPut(args map[string]any, by string) (any, error) { } else if found { if before, err := puts.Decode(raw); err == nil && !puts.Ended(before.Step) { if before.FP == fp && before.Hold == hold { - return putAnswer{Node: node, Module: module, Put: before, Follow: followPut(node, module)}, nil + a := answerOf(node, module, before) + a.Follow = followPut(node, module) + return a, nil } words += fmt.Sprintf("; it replaces the put of %s at %s, which stood at %s", module, shortFP(before.FP), before.Step) } @@ -182,7 +197,9 @@ func (s *putTools) doPut(args map[string]any, by string) (any, error) { return nil, fmt.Errorf("the put of %s on %s is not written: %w", module, node, err) } s.hear(key, data) - return putAnswer{Node: node, Module: module, Put: s.await(key, p), Follow: followPut(node, module)}, nil + a := answerOf(node, module, s.await(key, p)) + a.Follow = followPut(node, module) + return a, nil } // await is the put at key once the assigner's process took it up past its asking, or as it stands after answerWithin. @@ -248,7 +265,7 @@ func (s *putTools) doRelease(args map[string]any, by string) (any, error) { return nil, fmt.Errorf("the release of %s on %s is not written: %w", module, node, err) } s.hear(key, data) - return putAnswer{Node: node, Module: module, Put: p}, nil + return answerOf(node, module, p), nil } // doPuts is the tool assigner_puts. @@ -274,7 +291,7 @@ func (s *putTools) doPuts(args map[string]any) (any, error) { if err != nil { continue } - a := putAnswer{Node: node, Module: module, Put: p} + a := answerOf(node, module, p) if !puts.Ended(p.Step) { if !asked { asked = true diff --git a/modules/mesh-delivery-assigner/internal/puts/puts.go b/modules/mesh-delivery-assigner/internal/puts/puts.go index 06da4c03..ca70b3eb 100644 --- a/modules/mesh-delivery-assigner/internal/puts/puts.go +++ b/modules/mesh-delivery-assigner/internal/puts/puts.go @@ -97,6 +97,16 @@ func (p Put) Said(at time.Time, step, words string) Put { return p } +// StandsSince is when p came to stand at its step: the time of the first of its newest steps at that step (a step's words +// may change while it stands, each change a step of its own), or when it was asked when its steps say none. +func (p Put) StandsSince() time.Time { + since := p.At + for i := len(p.Steps) - 1; i >= 0 && p.Steps[i].Step == p.Step; i-- { + since = p.Steps[i].At + } + return since +} + var digestForm = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`) func (p Put) check() error {