From b3f7054345a7e290743a1be97c79f395841b701a Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 01:05:42 +0100 Subject: [PATCH] Builder: the operator's ask uses mesh-sdk's operatorask asker MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../cmd/mesh-delivery-builder/main.go | 12 ++++-- .../cmd/mesh-delivery-builder/retake.go | 41 +----------------- .../internal/build/retakeask.go | 19 +++++--- .../internal/build/retakeask_test.go | 43 +++++++++++++++++++ 4 files changed, 68 insertions(+), 47 deletions(-) diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go index cdba8de9..1bd7f2e7 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -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) } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/retake.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/retake.go index 90db6669..5c9cae8a 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/retake.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/retake.go @@ -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 `.` 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 { diff --git a/modules/mesh-delivery-builder/internal/build/retakeask.go b/modules/mesh-delivery-builder/internal/build/retakeask.go index c59f7bab..d5a0a8f9 100644 --- a/modules/mesh-delivery-builder/internal/build/retakeask.go +++ b/modules/mesh-delivery-builder/internal/build/retakeask.go @@ -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 }) } diff --git a/modules/mesh-delivery-builder/internal/build/retakeask_test.go b/modules/mesh-delivery-builder/internal/build/retakeask_test.go index 36a654d8..25cccb2c 100644 --- a/modules/mesh-delivery-builder/internal/build/retakeask_test.go +++ b/modules/mesh-delivery-builder/internal/build/retakeask_test.go @@ -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) + } +}