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

Builder: the operator's ask uses mesh-sdk's operatorask asker

Browse Source 
The builder's own lookup of the seat holder and read of the router's record (routerBucket, recordVia in
cmd/mesh-delivery-builder/retake.go) are deleted: the ask is published and the router's decided heard
through operatorask.Asker, and each warrant is held to its ask and acted on once by the operator-channel
client's Take.

Builds: design 47 G A-101 (one operator-channel asker in mesh-sdk), §700, A-74
Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
jschoubben
2026-11-12 01:05:42 +01:00
parent 62e56af786
commit b3f7054345
4 changed files with 68 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go
+9 -3
View File
@@ -19,6 +19,7 @@ import (
"syscall"
"time"
mma "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
@@ -36,6 +37,7 @@ import (
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
"git.novox.be/novox/mesh-sdk/go/operatorask"
"git.novox.be/novox/mesh-sdk/go/process"
"git.novox.be/novox/mesh-sdk/go/seatlist"
"git.novox.be/novox/mesh-sdk/go/views"
@@ -241,9 +243,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
// The operator is asked once a key's event builds are spent (design 47 G §568): the builder's asker on the seat operator-channel.
decided := &decisions{subject: oc.SubjectDecided(cred.Module)}
retakeActed := awaitedReserved{awaitedVersions{b: actedKV, bucket: versions.ActedBucket}}
operator := &build.OperatorAsk{Builder: b, Acted: retakeActed, Ctx: ctx, Logf: logger.Printf,
Client: &occ.Client{Module: cred.Module, State: calls.State("asked"), Publish: publishVia(ctx, js),
Record: recordVia(calls.Call, calls.Ask), Subscribe: decided.hear(logger.Printf)}}
asker := &operatorask.Asker[oc.Warrant]{Module: cred.Module, Call: calls.Call, Record: calls.Ask, Publish: publishVia(ctx, js), Subscribe: decided.hear(logger.Printf),
Contract: operatorask.Contract{Lists: "seat:" + mma.Seat + "." + mma.VerbAssignments, ListBody: mma.AssignmentsInput{Seats: mma.AssignmentsInputSeatsTrue},
Seat: oc.Seat, StateAsks: oc.StateAsks},
AskSubject: oc.AcceptSubjectAsk, DecidedSubject: oc.SubjectDecided}
operator := &build.OperatorAsk{Builder: b, Acted: retakeActed, Ctx: ctx, Logf: logger.Printf, Asker: asker,
Client: &occ.Client{Module: cred.Module, State: calls.State("asked"),
Publish: func(_ string, body any, id string) error { return asker.Ask(id, body) }}}
if err := operator.Listen(); err != nil {
return fmt.Errorf("hearing the operator's answers: %w", err)
}
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/retake.go
+2 -39
View File
@@ -3,10 +3,7 @@ package main
import (
"context"
"encoding/json"
"fmt"
mma "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
stdio "git.novox.be/novox/mesh-sdk/go"
"github.com/nats-io/nats.go/jetstream"
@@ -35,8 +32,8 @@ func (g *retaking) heard(ctx context.Context, m jetstream.Msg, logf func(string,
return false
}
// The builder's three bus adapters for the operator-channel client (design 47 G §568), as the teller has them: its own ask
// published, the router's record read through the seat, its own `decided` heard through the module's hub.
// The builder's two bus adapters for mesh-sdk's operator-channel asker (design 47 G §568, A-101): its own ask published and its
// own `decided` heard through the module's hub. The seat's holder and the router's record are the asker's.
// publishVia publishes an accept of the seat, de-duplicated by id: the same id twice is one message.
func publishVia(ctx context.Context, js jetstream.JetStream) func(subject string, body any, id string) error {
@@ -54,40 +51,6 @@ func publishVia(ctx context.Context, js jetstream.JetStream) func(subject string
}
}
// routerBucket is the bucket of the router's asks: the state `asks` of the seat operator-channel (the contract's StateAsks) in
// the bucket of the seat's holder on the seat list, which the seat mesh-module-assigner's verb assignments answers with seats
// (design 47 G A-74); no module's name is in the program.
func routerBucket(call func(key string, body any) (json.RawMessage, error)) (string, error) {
raw, err := call("seat:"+mma.Seat+"."+mma.VerbAssignments, mma.AssignmentsInput{Seats: mma.AssignmentsInputSeatsTrue})
if err != nil {
return "", fmt.Errorf("who holds the seat %s is not asked of the seat %s: %w", oc.Seat, mma.Seat, err)
}
var held struct {
Seats []mma.SeatHolders `json:"seats"`
}
if err := json.Unmarshal(raw, &held); err != nil {
return "", fmt.Errorf("the answer of %s.%s is not its seat's: %w", mma.Seat, mma.VerbAssignments, err)
}
for _, s := range held.Seats {
if s.Seat == oc.Seat && len(s.Holders) > 0 {
return s.Holders[0].Module + "_" + oc.StateAsks, nil
}
}
return "", fmt.Errorf("no module holds the seat %s on the seat list, so the router's record of an ask is not read", oc.Seat)
}
// recordVia reads the router's record of one of this module's asks: a get of `<module>.<id>` in the bucket the seat's holder
// keeps (record is process.Conn.Ask).
func recordVia(call func(key string, body any) (json.RawMessage, error), record func(bucket, id string) (*stdio.StateEntry, error)) func(id string) (*stdio.StateEntry, error) {
return func(id string) (*stdio.StateEntry, error) {
bucket, err := routerBucket(call)
if err != nil {
return nil, err
}
return record(bucket, id)
}
}
// decisions holds the handler of the router's `decided` event for this module: the module's one hub has the pattern
// `mesh.seat.*.event.>`, which covers it, so the events handler hands the message here (a hub refuses two patterns that overlap).
type decisions struct {
modules/mesh-delivery-builder/internal/build/retakeask.go
+14 -5
View File
@@ -10,6 +10,7 @@ import (
occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
"git.novox.be/novox/mesh-sdk/go/operatorask"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"mesh-delivery-builder/internal/versions"
@@ -34,9 +35,12 @@ const askBuildBound = 15 * time.Second
// OperatorAsk asks the operator about a key and carries out the answer.
type OperatorAsk struct {
// Client is the builder's asker on the seat operator-channel: its ask published under the builder's own name, its state
// `asked` (the asks it kept and the warrants it acted on), the router's record, the `decided` event.
// Client keeps the builder's asks and the warrants it acted on in its state `asked`, checks each ask and holds each
// warrant to the ask it answers; its Publish is Asker's.
Client *occ.Client
// Asker is mesh-sdk's asker on the seat operator-channel (design 47 G A-101): the ask published under the builder's own
// name and the router's `decided` heard; no copy of it is the builder's.
Asker *operatorask.Asker[oc.Warrant]
// Acted keeps which key an ask was made for, beside the revision of the key it was made at (Retaker.Acted).
Acted versions.ReserveStore
Builder *Builder
@@ -125,14 +129,19 @@ func (o *OperatorAsk) Ask(ctx context.Context, key string, cur builds.BuilderBui
return o.Client.Ask(a)
}
// Listen hears the operator's answers: each warrant of this builder's asks is acted on once.
// Listen hears the operator's answers through the sdk asker: each warrant of this builder's asks is acted on once.
func (o *OperatorAsk) Listen() error {
return o.Client.Listen(o.act, func(h occ.Handled, w oc.Warrant, err error) {
return o.Asker.Hear(func(w oc.Warrant) error {
h, err := o.Client.Take(w, o.act)
if h == "" && err != nil {
return err // the state could not be reached: offered again
}
if err != nil {
o.logf("the answer to the ask %s: %s: %v", w.Ask, h, err)
return
return nil
}
o.logf("the answer to the ask %s: %s", w.Ask, h)
return nil
})
}
modules/mesh-delivery-builder/internal/build/retakeask_test.go
+43
View File
@@ -9,6 +9,7 @@ import (
occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/operatorask"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
)
@@ -111,3 +112,45 @@ func levelOf(a oc.Ask, option string) string {
o, _ := occ.OptionOf(a, option)
return o.Level
}
// Design 47 G A-101: the operator's answer comes through mesh-sdk's asker (its `decided` subject for the builder), is held to
// the ask it answers and acted on once.
func TestTheOperatorsAnswerIsHeardThroughTheSdkAskerAndActedOnOnce(t *testing.T) {
r := newRig(t, goBundle)
r.consider(t, "modules/m")
seed(t, r.st, builds.BuilderBuild{State: builds.BuildFailed, By: "novox.builder", Manifest: mdigest('a'), EventBuilds: 3, Error: "transport fault"})
var hear func(stdio.Envelope) error
var subscribed string
var published []oc.Ask
state := &askedState{}
asker := &operatorask.Asker[oc.Warrant]{Module: "mesh-delivery-builder",
Subscribe: func(pattern string, h func(stdio.Envelope) error) error { subscribed, hear = pattern, h; return nil },
Publish: func(subject string, body any, _ string) error {
published = append(published, body.(oc.Ask))
return nil
},
AskSubject: oc.AcceptSubjectAsk, DecidedSubject: oc.SubjectDecided}
o := &OperatorAsk{Builder: r.b, Acted: &actedStore{}, Ctx: context.Background(), Asker: asker,
Now: func() time.Time { return time.Date(2026, 11, 11, 12, 0, 0, 0, time.UTC) },
Client: &occ.Client{Module: "mesh-delivery-builder", State: state, Publish: func(_ string, body any, id string) error { return asker.Ask(id, body) }}}
if err := o.Listen(); err != nil || subscribed != oc.SubjectDecided("mesh-delivery-builder") {
t.Fatalf("Listen = %v, subscribed to %q", err, subscribed)
}
key := r.key
if err := o.Ask(context.Background(), key, builds.BuilderBuild{State: builds.BuildFailed, EventBuilds: 3, Error: "transport fault"}, 7); err != nil || len(published) != 1 {
t.Fatalf("Ask = %v, published %d", err, len(published))
}
a := published[0]
w, _ := json.Marshal(oc.Warrant{Ask: a.ID, AskDigest: occ.Digest(a), Asker: "mesh-delivery-builder", Outcome: occ.OutcomeChosen, Option: OptionLeaveIt, Level: levelOf(a, OptionLeaveIt),
At: occ.FormatTime(time.Date(2026, 11, 11, 12, 5, 0, 0, time.UTC)), By: &oc.Person{Who: occ.Operator, Kind: "telegram", Identity: "op", Verified: "user"}})
for i := 0; i < 2; i++ { // given twice, acted on once
if err := hear(stdio.Envelope{Key: subscribed, Body: w}); err != nil {
t.Fatal(err)
}
}
var k occ.Kept
e, _ := state.Get("ask." + a.ID)
if err := json.Unmarshal(e.Value, &k); err != nil || !k.Acted || k.Answer == nil || k.Answer.Option != OptionLeaveIt {
t.Errorf("the ask as kept = %+v, %v; want acted on with the answer", k, err)
}
}