novox/mesh-delivery
19
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Assigner and grants part: record hand acts and warrants through the seat mesh-hand-acts, the controller's log only while that seat has no holder and the controller holds its lease (issue 1953 D8)

Browse Source 
Pins mesh-sdk 2d2aa7b in the L5 modules.
This commit is contained in:
jschoubben
2026-10-29 22:24:00 +01:00
parent 3b6f3af38e
commit 5235bb2e79
15 changed files with 90 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go
+3 -3
View File
@@ -387,7 +387,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
return
}
if bus != nil {
p.acts = newPutActs(bus, wake, logger.Printf)
p.acts = newPutActs(bus, wake, logger.Printf, w.handedAway)
}
logger.Printf("the bucket %s is open: puts are taken further", delivery.AssignerPutsBucket)
}
@@ -450,10 +450,10 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
p := &passer{kvs: kvs, w: w, logf: logger.Printf, say: say, said: map[string]bool{}, answeredAt: map[string]uint64{}}
p.asks = newComposeAsks(ctx, buildrun.Asker{Publisher: buildrun.JetStreamAsks{JS: js}, Waiter: buildrun.JetStreamAsks{JS: js}}, wake, logger.Printf)
if bus != nil && kvs[delivery.AssignerPutsBucket] != nil {
p.acts = newPutActs(bus, wake, logger.Printf)
p.acts = newPutActs(bus, wake, logger.Printf, w.handedAway)
}
if bus != nil {
p.seat = newSeatActs(bus, wake, logger.Printf)
p.seat = newSeatActs(bus, wake, logger.Printf, w.handedAway)
}
if js != nil {
p.publish = streamPublisher(ctx, js, thisNode)
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go
+2 -2
View File
@@ -187,7 +187,7 @@ func putOfTheBus(t *testing.T) {
p := newPasser(kvs, now)
bus := &fakeBus{calls: make(chan string, 8)}
woke := make(chan struct{}, 8) // each act's goroutine wakes a pass once it is done
p.acts = newPutActs(bus, func() { woke <- struct{}{} }, func(string, ...any) {})
p.acts = newPutActs(bus, func() { woke <- struct{}{} }, func(string, ...any) {}, nil)
p.acts.snapshot = func(context.Context, string, string) (string, error) {
return "novox's restore point of nats (streams) taken 22:01", nil
}
@@ -288,7 +288,7 @@ func putOfTheBus(t *testing.T) {
if got.Step != delivery.PutWanted || err != nil || w.FP != fp || w.Composed != digest || !w.Put || !w.Hold {
t.Fatalf("snapshotted, the put stands at %s (%s); the want %+v, %v", got.Step, got.Words, w, err)
}
if call := <-bus.calls; !strings.Contains(call, "warranted") || !strings.Contains(call, a.Ask.ID) {
if call := <-bus.calls; !strings.HasPrefix(call, "seat:mesh-hand-acts.warranted ") || !strings.Contains(call, a.Ask.ID) { // ADR 0565 §11: into the seat
t.Errorf("the warrant was recorded as %q", call)
}
world(delivery.AssignerWantsBucket, "want.novox.nats", wantsKV.values["want.novox.nats"])
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go
+10 -6
View File
@@ -13,6 +13,7 @@ import (
"git.novox.be/novox/mesh-sdk/go/asks"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/handacts"
"mesh-delivery-assigner/internal/assign"
)
@@ -46,14 +47,16 @@ type putActs struct {
snapped map[string]string
snapNo map[string]string
warranted map[string]bool
// rec records the put's warrant in the hand-act log (ADR 0565 §11).
rec handacts.Recorder
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{}, warrantNo: map[string]string{},
func newPutActs(bus caller, wake func(), logf func(string, ...any), leaseAway func() bool) *putActs {
a := &putActs{rec: handacts.Recorder{Ask: bus.Call, LeaseAway: leaseAway}, 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
@@ -381,8 +384,9 @@ func (a *putActs) snapshotTheBus(ctx context.Context, node, module string) (stri
}
}
// warrant records once, in the controller's hand-act log, that the bus's want was written on the operator's warrant: the
// router's record of the teller's ask says who chose and how (ADR 0274).
// warrant records once, in the hand-act log (the seat mesh-hand-acts', or the controller's while that seat has no holder and
// the controller holds its lease, ADR 0565 §11), that the bus's want was written on the operator's warrant: the router's
// record of the teller's ask says who chose and how (ADR 0274).
func (a *putActs) warrant(id, ask string) {
a.mu.Lock()
if a.warranted[id] {
@@ -392,10 +396,10 @@ func (a *putActs) warrant(id, ask string) {
a.warranted[id] = true
a.mu.Unlock()
go func() {
if _, err := a.bus.Call("seat:"+catalogue.ControllerSeatName+".warranted", map[string]any{"asker": delivery.TellerModule, "ask": ask}); err != nil {
if _, err := a.rec.RecordWarrant(delivery.TellerModule, 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.warrantNo[id] = fmt.Sprintf("the warrant %s is not recorded in the hand-act log: %v", ask, err)
a.mu.Unlock()
a.wake()
}
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go
+17 -8
View File
@@ -21,6 +21,7 @@ import (
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
"git.novox.be/novox/mesh-sdk/go/compose/judge"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/handacts"
"git.novox.be/novox/mesh-sdk/go/process"
"mesh-delivery-assigner/internal/assign"
@@ -67,6 +68,8 @@ type seatActs struct {
export map[string]exportCheck
// calls is, by pair (node and module, or node and provision for a pin), what the last call for it came to.
calls map[string]string
// rec records the seat's acts in the hand-act log (ADR 0565 §11).
rec handacts.Recorder
}
// exportCheck is the export read again for one proposal: running, its answer, or why it was not read.
@@ -76,8 +79,9 @@ type exportCheck struct {
err string
}
func newSeatActs(bus caller, wake func(), logf func(string, ...any)) *seatActs {
return &seatActs{bus: bus, wake: wake, logf: logf, export: map[string]exportCheck{}, calls: map[string]string{}}
func newSeatActs(bus caller, wake func(), logf func(string, ...any), leaseAway func() bool) *seatActs {
return &seatActs{bus: bus, wake: wake, logf: logf, export: map[string]exportCheck{}, calls: map[string]string{},
rec: handacts.Recorder{Ask: bus.Call, LeaseAway: leaseAway}}
}
// exportOf is the export read again for a proposal, asked once in the background.
@@ -114,26 +118,31 @@ func (a *seatActs) forget(id string) {
a.mu.Unlock()
}
// record records once, in the controller's hand-act log, that placement was written on the operator's warrant (§5): the
// router's record of the teller's ask says who chose and how (ADR 0274).
// record records once, in the hand-act log, that placement was written on the operator's warrant (§5): the router's
// record of the teller's ask says who chose and how (ADR 0274). The log is the seat mesh-hand-acts' from its delivery, and
// the controller's while that seat has no holder and the controller holds its lease (ADR 0565 §11, handacts.Recorder).
func (a *seatActs) record(warrant string) {
if a == nil || warrant == "" {
return
}
ask, _, _ := strings.Cut(warrant, "@")
go func() {
if _, err := a.bus.Call("seat:"+catalogue.ControllerSeatName+".warranted", map[string]any{"asker": delivery.TellerModule, "ask": ask}); err != nil {
if taken, err := a.rec.RecordWarrant(delivery.TellerModule, ask); err != nil {
a.logf("the placement's warrant %s is not recorded in the hand-act log: %v", ask, err)
} else {
a.logf("the placement's warrant %s is recorded in the log of %s", ask, taken.Log)
}
}()
}
// handAct records a hand-back in the controller's hand-act log (§7).
// handAct records a hand-back in the hand-act log (§7), as record does.
func (a *seatActs) handAct(prop delivery.PlacementProposal) {
go func() {
if _, err := a.bus.Call(handActCall, map[string]any{"what": "mesh-placement hand-back of " + strings.Join(prop.Nodes(), ", "),
"why": prop.Why, "cause": "asked by " + prop.By + ", approved by the operator"}); err != nil {
if taken, err := a.rec.Record(handacts.Request{What: "mesh-placement hand-back of " + strings.Join(prop.Nodes(), ", "),
Why: prop.Why, Cause: "asked by " + prop.By + ", approved by the operator"}); err != nil {
a.logf("the hand-back %s is not recorded in the hand-act log: %v", prop.ID, err)
} else {
a.logf("the hand-back %s is recorded in the log of %s", prop.ID, taken.Log)
}
}()
}
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools.go
-3
View File
@@ -318,9 +318,6 @@ func (s *seatTools) leased() bool {
return away
}
// handActCall records a hand act in the controller's hand-act log.
const handActCall = "seat:" + catalogue.ControllerSeatName + ".hand-act"
// doHandBack is the verb hand-back (§7): the operator's alone, so it is asked as a proposal is. It writes a hand-back
// proposal naming every key at its revision; the assigner raises its authorising ask and, on its warrant, deletes the keys,
// records the hand act and says placement-changed; every node returns to the controller's compositions.