From 4b503f8d0d6a5a6aabe503700bab68050c925847 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 15:29:05 +0200 Subject: [PATCH] Pull a build seat's ask through the bundle's runtime one at a time and settle it by id, so a tools bundle pulls by its capacity and order (issue 1887 A11) --- go/buildstep/pull_seat.go | 44 +++++++++++++++++++++++ go/buildstep/pull_seat_test.go | 40 +++++++++++++++++++++ go/seat.go | 55 ++++++++++++++++++++++++++++ go/seat_pull_test.go | 65 ++++++++++++++++++++++++++++++++++ 4 files changed, 204 insertions(+) create mode 100644 go/buildstep/pull_seat.go create mode 100644 go/buildstep/pull_seat_test.go create mode 100644 go/seat_pull_test.go diff --git a/go/buildstep/pull_seat.go b/go/buildstep/pull_seat.go new file mode 100644 index 0000000..1a8680f --- /dev/null +++ b/go/buildstep/pull_seat.go @@ -0,0 +1,44 @@ +package buildstep + +import ( + "context" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// SeatAsks is a build seat's workers as a tools bundle reaches them: through its runtime's seat pull and settle +// (stdio.SeatPull, SeatSettle; issue 1887 A11), one message per pull, only from the workers the module's membership +// lists. +type SeatAsks struct { + Language string + // Pull and Settle are stdio.SeatPull and stdio.SeatSettle when nil; a test gives its own. + Pull func(worker string, wait time.Duration) (*stdio.Pulled, error) + Settle func(id, fate string) error +} + +// Next is AskSource's. +func (s SeatAsks) Next(ctx context.Context, kind string, wait time.Duration) (*Pulled, error) { + pull, settle := s.Pull, s.Settle + if pull == nil { + pull = stdio.SeatPull + } + if settle == nil { + settle = stdio.SeatSettle + } + worker, err := delivery.AskWorker(s.Language, kind) + if err != nil { + return nil, err + } + if err := ctx.Err(); err != nil { + return nil, err + } + p, err := pull(worker, wait) + if err != nil || p == nil { + return nil, err + } + fate := func(f string) func() error { return func() error { return settle(p.ID, f) } } + return &Pulled{Data: p.Body, Delivered: p.Delivered, Ack: fate(stdio.SettleAck), InProgress: fate(stdio.SettleProgress), + Nak: fate(stdio.SettleNak), Term: fate(stdio.SettleTerm)}, nil +} diff --git a/go/buildstep/pull_seat_test.go b/go/buildstep/pull_seat_test.go new file mode 100644 index 0000000..7d70782 --- /dev/null +++ b/go/buildstep/pull_seat_test.go @@ -0,0 +1,40 @@ +package buildstep + +import ( + "context" + "testing" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// 1887/A11: a bundle's source pulls the kind's worker by its seat's name and settles each ask by the runtime's id. +func TestSeatAsksPullsTheKindsWorkerAndSettlesByID(t *testing.T) { + var pulled []string + var settled []string + src := SeatAsks{Language: delivery.AskGo, + Pull: func(worker string, wait time.Duration) (*stdio.Pulled, error) { + pulled = append(pulled, worker) + if worker != "SEAT_NODE_BUILD_GO_CHECK_worker" { + return nil, nil + } + return &stdio.Pulled{ID: "p9", Worker: worker, Body: []byte(`{"v":1}`), Delivered: 2}, nil + }, + Settle: func(id, fate string) error { settled = append(settled, id+" "+fate); return nil }} + if p, err := src.Next(context.Background(), delivery.AskDelivery, 0); err != nil || p != nil { + t.Fatalf("delivery = %+v, %v", p, err) + } + p, err := src.Next(context.Background(), delivery.AskCheck, time.Second) + if err != nil || p == nil || string(p.Data) != `{"v":1}` || p.Delivered != 2 { + t.Fatalf("check = %+v, %v", p, err) + } + _ = p.InProgress() + _ = p.Ack() + if len(pulled) != 2 || pulled[0] != "SEAT_NODE_BUILD_GO_DELIVERY_worker" { + t.Errorf("pulled %v", pulled) + } + if len(settled) != 2 || settled[0] != "p9 progress" || settled[1] != "p9 ack" { + t.Errorf("settled %v", settled) + } +} diff --git a/go/seat.go b/go/seat.go index aa2f144..18f49a5 100644 --- a/go/seat.go +++ b/go/seat.go @@ -69,6 +69,61 @@ func SeatTake(worker string, handle func(Work) error) error { return err } +// Pulled is one message a bundle pulled from a seat's worker (SeatPull), named by the runtime's id for it, which +// SeatSettle gives its fate. +type Pulled struct { + ID string `json:"id"` + Worker string `json:"worker"` + Subject string `json:"subject"` + Body json.RawMessage `json:"body"` + Headers map[string]string `json:"headers,omitempty"` + Delivered uint64 `json:"delivered"` +} + +// The fates SeatSettle gives a pulled message: done, handed back to be delivered again, dropped, or still being worked. +const ( + SettleAck = "ack" + SettleNak = "nak" + SettleTerm = "term" + SettleProgress = "progress" +) + +// SeatPull pulls at most one message from a held seat's worker the module's membership lists, waiting at most wait +// (none when 0, at most 25 s): nil when none came. Unlike SeatTake nothing is fetched ahead: a build module pulls an +// ask only when it has a free slot (novox/hq ADR 0525 ยง4; issue 1887 A11). A message never settled is delivered again +// by the worker's acknowledgement wait. +func SeatPull(worker string, wait time.Duration) (*Pulled, error) { + c, err := live() + if err != nil { + return nil, err + } + raw, err := c.ask("mesh/seat.pull", map[string]any{"worker": worker, "wait_ms": wait.Milliseconds()}, seatTimeout) + if err != nil { + return nil, err + } + if len(raw) == 0 || string(raw) == "null" { + return nil, nil + } + var p Pulled + if err := json.Unmarshal(raw, &p); err != nil { + return nil, err + } + if p.ID == "" { + return nil, nil + } + return &p, nil +} + +// SeatSettle gives the message pulled under id its fate (SettleAck, SettleNak, SettleTerm, SettleProgress). +func SeatSettle(id, fate string) error { + c, err := live() + if err != nil { + return err + } + _, err = c.ask("mesh/seat.settle", map[string]any{"id": id, "fate": fate}, seatTimeout) + return err +} + // SeatRecord reads one key of a record a seat keeps for this module, under its own name, or nil. func SeatRecord(bucket, key string) (*StateEntry, error) { c, err := live() diff --git a/go/seat_pull_test.go b/go/seat_pull_test.go new file mode 100644 index 0000000..da7fbe5 --- /dev/null +++ b/go/seat_pull_test.go @@ -0,0 +1,65 @@ +package stdio + +import ( + "bufio" + "encoding/json" + "io" + "testing" + "time" +) + +// 1887/A11: a bundle pulls one message through its runtime, naming the worker and the wait, and settles it by the id the +// runtime gave. +func TestABundlePullsAndSettlesThroughTheRuntime(t *testing.T) { + t.Setenv(ServedAs, "mesh-build-go") + inR, inW := io.Pipe() + outR, outW := io.Pipe() + got := make(chan *Pulled, 1) + go func() { + _ = ServeOn("", []Tool{{Name: "start", Run: func(map[string]any) (any, error) { + p, err := SeatPull("SEAT_NODE_BUILD_GO_DELIVERY_worker", 1500*time.Millisecond) + if err != nil { + return nil, err + } + got <- p + return map[string]any{}, SeatSettle(p.ID, SettleAck) + }}}, inR, outW) + }() + lines := bufio.NewScanner(outR) + next := func() map[string]any { + done := make(chan map[string]any, 1) + go func() { + if lines.Scan() { + var m map[string]any + _ = json.Unmarshal(lines.Bytes(), &m) + done <- m + } + }() + select { + case m := <-done: + return m + case <-time.After(3 * time.Second): + t.Fatal("the bundle said nothing") + return nil + } + } + send := func(m map[string]any) { b, _ := json.Marshal(m); _, _ = inW.Write(append(b, '\n')) } + send(map[string]any{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": map[string]any{"name": "start"}}) + asked := next() + params, _ := asked["params"].(map[string]any) + if asked["method"] != "mesh/seat.pull" || params["worker"] != "SEAT_NODE_BUILD_GO_DELIVERY_worker" || params["wait_ms"] != 1500.0 { + t.Fatalf("asked %v", asked) + } + send(map[string]any{"jsonrpc": "2.0", "id": asked["id"], "result": map[string]any{"id": "p1", "worker": "SEAT_NODE_BUILD_GO_DELIVERY_worker", + "subject": "mesh.seat.node-build-go.accept.delivery", "body": map[string]any{"v": 1}, "delivered": 1}}) + p := <-got + if p.ID != "p1" || string(p.Body) != `{"v":1}` { + t.Fatalf("pulled %+v", p) + } + asked = next() + params, _ = asked["params"].(map[string]any) + if asked["method"] != "mesh/seat.settle" || params["id"] != "p1" || params["fate"] != "ack" { + t.Fatalf("settled %v", asked) + } + send(map[string]any{"jsonrpc": "2.0", "id": asked["id"], "result": map[string]any{}}) +}