diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go index 71b88a1..0458ebc 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go @@ -305,6 +305,36 @@ func putOfTheBus(t *testing.T) { if len(bus.events) < 5 || !strings.Contains(strings.Join(bus.events, "\n"), "holding") { t.Errorf("the steps said as events: %v", bus.events) } + // The same bus put again at the same composition after the release is its own asking: its ask is written again under an + // id of its own, and the earlier yes, still at the answers' key, does not count for it (ADR 0546 clause 8). + world(delivery.HaveBucket, hkey, before) + s.now = func() time.Time { return now.Add(time.Minute) } + if _, err := s.doPut(map[string]any{"node": "novox", "module": "nats", "version": fp, "hold": "true", "why": "again"}, "shanks.node-tools"); err != nil { + t.Fatal(err) + } + again := step() + a2, err := delivery.DecodeAssignerAsk(akey, asksKV.values[akey]) + if again.Step != delivery.PutPerson || err != nil || a2.Ask.ID == a.Ask.ID { + t.Fatalf("put again, it stands at %s (%s); its ask %q, %v; the first ask was %q", again.Step, again.Words, a2.Ask.ID, err, a.Ask.ID) + } + world(delivery.AssignerAsksBucket, akey, asksKV.values[akey]) + yes2, _ := delivery.EncodeAnswer(delivery.Answer{Option: assign.OptionYes, Warrant: a2.Ask.ID + "@2026-10-25T22:03:00Z", Words: "the operator, verified"}) + world(delivery.AnswersBucket, akey, yes2) + if got := step(); got.Step != delivery.PutSnapshot { + t.Fatalf("on the second warrant the put stands at %s: %s", got.Step, got.Words) + } + <-woke + // Released between the pass's read and its write: the pass writes no want for the put that no longer stands. + oldWant, _ := delivery.EncodeWant(delivery.Want{FP: delivery.SourcePrefix2 + strings.Repeat("a", 64)}) + wantsKV.values["want.novox.nats"] = oldWant + world(delivery.AssignerWantsBucket, "want.novox.nats", oldWant) + if _, err := s.doRelease(map[string]any{"node": "novox", "module": "nats", "why": "changed my mind"}, "shanks.node-tools"); err != nil { + t.Fatal(err) + } + p.pass(ctx) + if string(wantsKV.values["want.novox.nats"]) != string(oldWant) { + t.Errorf("a want was written for a put released meanwhile: %s", wantsKV.values["want.novox.nats"]) + } } // 1945/AC21: given a module on neither list on a node, when passes follow, then the assigner raises its condition once, says it 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 c5216be..7549bb9 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go @@ -10,6 +10,7 @@ import ( "sync" "time" + "git.novox.be/novox/mesh-sdk/go/asks" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" @@ -45,13 +46,14 @@ type putActs struct { snapped map[string]string snapNo map[string]string warranted map[string]bool + warrantNo map[string]string said map[string]string // 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) } func newPutActs(bus caller, wake func(), logf func(string, ...any)) *putActs { - a := &putActs{bus: bus, wake: wake, logf: logf, snapping: map[string]bool{}, snapped: map[string]string{}, snapNo: map[string]string{}, warranted: map[string]bool{}, + a := &putActs{bus: bus, wake: wake, logf: logf, snapping: map[string]bool{}, snapped: map[string]string{}, snapNo: map[string]string{}, warranted: map[string]bool{}, warrantNo: map[string]string{}, said: map[string]string{}} a.snapshot = a.snapshotTheBus return a @@ -125,9 +127,20 @@ func (p *passer) puts(ctx context.Context) { next.Put.Composed = w.Composed } } + // The bus's want is written only for the act the operator's yes bound (ADR 0546 clause 8). + if next.Put.Step == delivery.PutWanted && put.Step != delivery.PutWanted && world.Bus && next.Put.Ask != "" { + if err := p.performs(next.Put.Ask, assign.PutAct(node, module, next.Put.FP, next.Put.Composed)); err != nil { + next = assign.PutNext{Put: next.Put.Said(world.Now, delivery.PutRefused, "the bus's want is not written: "+err.Error())} + } + } + if failed := p.acts.warrantFailed(id); failed != "" && (next.Put.Step == delivery.PutWanted || next.Put.Step == delivery.PutHolding) && + !strings.Contains(next.Put.Words, failed) { + next.Put = next.Put.Said(world.Now, next.Put.Step, next.Put.Words+"; "+failed) + } p.actOn(ctx, node, module, id, world, next) - if !samePut(put, next.Put) { - p.writePut(ctx, key, put, next.Put) + written := samePut(put, next.Put) || p.writePut(ctx, key, put, next.Put) + if !written { + continue // released or put again meanwhile: nothing is done for the put that no longer stands } if next.Put.Step == delivery.PutWanted && put.Step != delivery.PutWanted { // The want is written in this pass too, before the next one would: Module.Puts writes the same. @@ -137,6 +150,9 @@ func (p *passer) puts(ctx context.Context) { p.put(ctx, delivery.AssignerWantsBucket, wkey, func() ([]byte, error) { return delivery.EncodeWant(want) }) } } + if next.Warranted { + p.acts.warrant(id, delivery.AskingID(next.Put.Ask, assign.PutAsking(next.Put))) + } } } } @@ -172,11 +188,16 @@ func (p *passer) putWorld(node, module string, put delivery.Put, manifest catalo } w.Recorded = policy == catalogue.PolicyRecord if akey, err := delivery.RecordAskKey(node, module, d.Digest); err == nil && put.Ask == akey { - switch { - case p.answeredWith(akey, assign.OptionYes): - w.Answer = assign.OptionYes - case p.answeredWith(akey, assign.OptionNo): - w.Answer = assign.OptionNo + // Only this put's own ask counts: an ask (and its answer) of an earlier put under the same key is another asking. + if a, err := delivery.DecodeAssignerAsk(akey, p.w.value(delivery.AssignerAsksBucket, akey)); err == nil && a.Asking == assign.PutAsking(put) { + switch { + case p.answeredWith(akey, assign.OptionYes): + w.Answer = assign.OptionYes + case p.answeredWith(akey, assign.OptionNo): + w.Answer = assign.OptionNo + default: + w.AskExpired = !a.Ask.Expires.IsZero() && w.Now.After(a.Ask.Expires) + } } } } @@ -212,9 +233,12 @@ func (p *passer) composedFor(node, module, fp string) (*delivery.ComposedDigest, // actOn does the acts a step asks for. func (p *passer) actOn(ctx context.Context, node, module, id string, w assign.PutWorld, next assign.PutNext) { - if next.Ask && w.Composed != nil { + if next.Put.Step == delivery.PutPerson && next.Put.Ask != "" && w.Composed != nil { + // The put's own ask, written again over one an earlier put left under the key (its asking differs), so the teller + // raises it anew under its own id. akey := next.Put.Ask - if !p.w.has(delivery.AssignerAsksBucket, akey) { + current, err := delivery.DecodeAssignerAsk(akey, p.w.value(delivery.AssignerAsksBucket, akey)) + if err != nil || current.Asking != assign.PutAsking(next.Put) { a, err := assign.PutAsk(akey, node, module, next.Put, *w.Composed, w.Bus, w.Now) if err != nil { p.logf("the ask about putting %s on %s is not written: %v", module, node, err) @@ -226,9 +250,20 @@ func (p *passer) actOn(ctx context.Context, node, module, id string, w assign.Pu if next.Snapshot { p.acts.snap(ctx, id, node, module) } - if next.Warranted { - p.acts.warrant(id, delivery.AskingID(next.Put.Ask, 0)) +} + +// performs checks that the act about to be performed is the one the yes of the ask at key bound (asks.Option.Performs). +func (p *passer) performs(key string, act asks.Act) error { + a, err := delivery.DecodeAssignerAsk(key, p.w.value(delivery.AssignerAsksBucket, key)) + if err != nil { + return fmt.Errorf("the ask %s cannot be read: %w", key, err) } + for _, o := range a.Ask.Options { + if o.ID == assign.OptionYes { + return o.Performs(act) + } + } + return fmt.Errorf("the ask %s offers no yes", key) } // samePut says two puts say the same: their step, words and what they found. @@ -237,35 +272,36 @@ func samePut(a, b delivery.Put) bool { } // writePut writes a put's next value over the one the pass read, by compare-and-set, unless the tool has put again or -// released it since; and says it as the event put. -func (p *passer) writePut(ctx context.Context, key string, read, next delivery.Put) { +// released it since; and says it as the event put. It says whether it wrote. +func (p *passer) writePut(ctx context.Context, key string, read, next delivery.Put) bool { kv := p.kvs[delivery.AssignerPutsBucket] if kv == nil { - return + return false } e, err := kv.Get(ctx, key) if err != nil { p.logf("the put %s cannot be read again, and is taken further at the next pass: %v", key, err) - return + return false } now, err := delivery.DecodePut(e.Value()) if err != nil || !now.At.Equal(read.At) || now.Step != read.Step { - return // asked again or released meanwhile: the next pass reads that + return false // asked again or released meanwhile: the next pass reads that } data, err := delivery.EncodePut(next) if err != nil { p.logf("the put %s is not written: %v", key, err) - return + return false } if _, err := kv.Update(ctx, key, data, e.Revision()); err != nil { if !errors.Is(err, context.Canceled) { p.logf("the put %s could not be written, and is taken further at the next pass: %v", key, err) } - return + return false } p.w.put(delivery.AssignerPutsBucket, key, data) p.logf("put %s: %s: %s", key, next.Step, next.Words) p.acts.say(key, next) + return true } // state is what this process knows of a put's acts. @@ -359,10 +395,21 @@ func (a *putActs) warrant(id, ask string) { go func() { if _, err := a.bus.Call("seat:"+catalogue.ControllerSeatName+".warranted", map[string]any{"asker": delivery.TellerModule, "ask": ask}); err != nil { a.logf("the put's warrant %s is not recorded in the hand-act log: %v", ask, err) + a.mu.Lock() + a.warrantNo[id] = fmt.Sprintf("the warrant %s is not recorded in the controller's hand-act log: %v", ask, err) + a.mu.Unlock() + a.wake() } }() } +// warrantFailed says why the put's warrant was not recorded, "" when it was or is not yet known. +func (a *putActs) warrantFailed(id string) string { + a.mu.Lock() + defer a.mu.Unlock() + return a.warrantNo[id] +} + // putSaid is the body of the event put. type putSaid struct { Node string `json:"node"` diff --git a/modules/mesh-delivery-assigner/internal/assign/asks_test.go b/modules/mesh-delivery-assigner/internal/assign/asks_test.go index 1fa82b1..0f0fbdb 100644 --- a/modules/mesh-delivery-assigner/internal/assign/asks_test.go +++ b/modules/mesh-delivery-assigner/internal/assign/asks_test.go @@ -171,6 +171,18 @@ func TestTheRecordAndBackupAsksAreValid(t *testing.T) { case !bus && (yes.Level != asks.Acknowledge || viewed != nil): t.Errorf("a recorded module: the yes is %+v, the view's answer %v", yes, viewed) } + later := put + later.At = now.Add(time.Minute) + b, _ := PutAsk(rkey, "ace", "mail", later, d, bus, now) + if b.Ask.ID == a.Ask.ID || b.Asking != PutAsking(later) { + t.Errorf("bus %v: a later put's ask is asking %d under id %q, the earlier's %q", bus, b.Asking, b.Ask.ID, a.Ask.ID) + } + if err := yes.Performs(PutAct("ace", "mail", fNew, dgst)); err != nil { + t.Errorf("bus %v: the yes does not perform the act it bound: %v", bus, err) + } + if err := yes.Performs(PutAct("ace", "mail", fNew, dig(fOld))); bus && err == nil { + t.Errorf("bus: the yes performs another composition's act") + } } } diff --git a/modules/mesh-delivery-assigner/internal/assign/put.go b/modules/mesh-delivery-assigner/internal/assign/put.go index e6f13c8..6a4d283 100644 --- a/modules/mesh-delivery-assigner/internal/assign/put.go +++ b/modules/mesh-delivery-assigner/internal/assign/put.go @@ -32,8 +32,10 @@ type PutWorld struct { // none. Granted says the grant at the fingerprint names the node's accounts. Both are read, never asked for (clause 4). Composed *delivery.ComposedDigest Granted bool - // Answer is the option the operator chose on the put's ask for the composition, "" when none. - Answer string + // Answer is the option the operator chose on this put's own ask (its asking, PutAsking), "" when none; AskExpired says + // that ask expired unanswered. + Answer string + AskExpired bool // Snapshot says where the bus's streams were snapshotted for this put, once they were; Snapshotting that it was // asked; SnapshotFailed why it was not taken. Snapshot string @@ -108,6 +110,9 @@ func StepPut(w PutWorld) PutNext { return say(delivery.PutRefused, "the operator answered no to putting %s at %s on %s", w.Module, fp, w.Node) case OptionYes: default: + if w.AskExpired { + return say(delivery.PutRefused, "the operator's ask about putting %s at %s on %s expired unanswered: the node keeps what it runs", w.Module, fp, w.Node) + } next := say(delivery.PutPerson, "waits for the operator's yes to putting %s at %s on %s (composition %s, judged %s)", w.Module, fp, w.Node, short(d.Digest), d.Judgement) if p.Ask == "" { next.Ask = true @@ -194,6 +199,21 @@ func PutOwns(p delivery.Put) (want *delivery.Want, owns bool) { return nil, true } +// PutAsking is a put's own asking number: when it was asked, in milliseconds. Each put asked under the same key is another +// asking, so the ask is written again under an id of its own and an answer to an earlier put, whose warrant names that put's +// asking, never counts for a later one (ADR 0546 clause 8: one approval binds one act). +func PutAsking(p delivery.Put) int { + if n := p.At.UnixMilli(); n > 0 { + return int(n) + } + return 0 +} + +// PutAct is the act a bus put's yes binds and that is performed when its want is written. +func PutAct(node, module, fp, composed string) asks.Act { + return asks.Act{"put": "want", "node": node, "module": module, "fp": fp, "composed": composed} +} + // PutAsk is the ask about a put's move (ADR 0546 clauses 7 and 8), under the record ask's key of the composition so the // teller relays it and its answer as any record ask's. For the bus its yes is of level approve and binds the act, so only a // channel that proves who answered takes it (ADR 0259), never the view. @@ -209,8 +229,7 @@ func PutAsk(key, node, module string, p delivery.Put, d delivery.ComposedDigest, explanation := fmt.Sprintf("Asked to put %s at version %s on %s: %s. The move %s.", module, short(p.FP), node, p.Why, what) expires := now.Add(7 * 24 * time.Hour) if bus { - act := asks.Act{"put": "want", "node": node, "module": module, "fp": p.FP, "composed": d.Digest} - binds, err := asks.ActDigest(act) + binds, err := asks.ActDigest(PutAct(node, module, p.FP, d.Digest)) if err != nil { return delivery.AssignerAsk{}, err } @@ -220,7 +239,7 @@ func PutAsk(key, node, module string, p delivery.Put, d delivery.ComposedDigest, expires = now.Add(asks.ApproveLasts) } return delivery.AssignerAsk{Ask: asks.Ask{ - ID: delivery.AskingID(key, 0), + ID: delivery.AskingID(key, PutAsking(p)), Headline: fmt.Sprintf("Put %s on %s?", module, node), Explanation: explanation, Options: []asks.Option{yes, @@ -229,5 +248,5 @@ func PutAsk(key, node, module string, p delivery.Put, d delivery.ComposedDigest, Expires: expires, OnExpiry: "the node keeps what it runs", About: key, - }}, nil + }, Asking: PutAsking(p)}, nil } diff --git a/modules/mesh-delivery-assigner/internal/assign/wants_test.go b/modules/mesh-delivery-assigner/internal/assign/wants_test.go index f1a3684..7c213b4 100644 --- a/modules/mesh-delivery-assigner/internal/assign/wants_test.go +++ b/modules/mesh-delivery-assigner/internal/assign/wants_test.go @@ -459,6 +459,7 @@ func stepsOfAPut(t *testing.T) { }, delivery.PutRefused, "mesh-hand-compose", false, false, false}, {"the bus asks the operator", bus, delivery.PutPerson, "waits for the operator's yes", true, false, false}, {"the operator said no", func() PutWorld { w := bus(); w.Answer = OptionNo; return w }, delivery.PutRefused, "answered no", false, false, false}, + {"the ask expired unanswered", func() PutWorld { w := bus(); w.AskExpired = true; return w }, delivery.PutRefused, "expired unanswered", false, false, false}, {"yes: the snapshot is asked", func() PutWorld { w := bus(); w.Answer = OptionYes; return w }, delivery.PutSnapshot, "asked novox's backup holder", false, true, false}, {"the snapshot is taken", func() PutWorld { w := bus(); w.Answer, w.Snapshotting = OptionYes, true; return w }, delivery.PutSnapshot, "waits for novox's backup holder", false, false, false}, {"the snapshot failed", func() PutWorld { w := bus(); w.Answer, w.SnapshotFailed = OptionYes, "no restic"; return w }, delivery.PutRefused, "no restic", false, false, false},