Let a bundle pull one message from its listed worker and settle it, so a build module takes an ask only when it has a free slot, in its workers' order (issue 1887 A11, ADR 0525 §4)

node-tools/internal/bus/seat_pull.go

@@ -0,0 +1,181 @@
+package bus
+
+import (
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// **A holder that pulls one piece of work when it has room** (novox/hq ADR 0525 §4, §6; issue 1887 A11). SeatTake
+// hands a bundle every piece a worker holds, one at a time and prefetched; a build module must instead take one ask
+// only when it has a free slot, from its workers in their order (delivery, then check, then recheck), and decide each
+// ask's fate when its step ends. SeatPull is one fetch of at most one message from a worker the module's membership
+// lists, never a prefetch; SeatSettle gives a pulled message its fate. A message the bundle never settles (its process
+// died) is delivered again by the worker's own acknowledgement wait.
+
+// MaxPullWait bounds how long one pull waits for a message: under the bundle's 30 s wait for its answer.
+var MaxPullWait = 25 * time.Second
+
+// Pulled is one message a module pulled from a seat's worker, named by an id the runtime gave it.
+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"`
+}
+
+// pulls are the messages modules pulled and have not settled, by id; each kept with the module it was pulled for, so
+// one module never settles another's.
+type pulls struct {
+	mu   sync.Mutex
+	held map[string]pulledMsg
+	cons map[string]jetstream.Consumer // by module and consumer
+}
+
+type pulledMsg struct {
+	module string
+	msg    jetstream.Msg
+}
+
+var seatPulls = &pulls{held: map[string]pulledMsg{}, cons: map[string]jetstream.Consumer{}}
+
+func pullID() string {
+	var b [8]byte
+	_, _ = rand.Read(b[:])
+	return hex.EncodeToString(b[:])
+}
+
+// SeatPull fetches at most one message from consumer for module, waiting at most wait (none when 0): nil when none
+// came. The consumer must be one the module's membership lists as its worker (novox/hq ADR 0259); the runtime binds to
+// it by name and never makes one.
+func (c *Conn) SeatPull(module, consumer string, wait time.Duration) (*Pulled, error) {
+	t, err := c.trafficOf(module)
+	if err != nil {
+		return nil, err
+	}
+	var w *SeatWorker
+	var names []string
+	for i := range t.Workers {
+		names = append(names, t.Workers[i].Consumer)
+		if t.Workers[i].Consumer == consumer {
+			w = &t.Workers[i]
+		}
+	}
+	if w == nil {
+		return nil, fmt.Errorf("%s pulls no work from %s: its membership lists %s (novox/hq ADR 0259)", module,
+			consumer, strings.Join(names, ", "))
+	}
+	if wait > MaxPullWait {
+		wait = MaxPullWait
+	}
+	bound, err := c.pullConsumer(module, *w)
+	if err != nil {
+		return nil, err
+	}
+	var batch jetstream.MessageBatch
+	if wait <= 0 {
+		batch, err = bound.FetchNoWait(1)
+	} else {
+		batch, err = bound.Fetch(1, jetstream.FetchMaxWait(wait))
+	}
+	if err != nil {
+		return nil, fmt.Errorf("pulling %s's worker %s: %w", module, w.Consumer, err)
+	}
+	msg, ok := <-batch.Messages()
+	if !ok {
+		if err := batch.Error(); err != nil {
+			return nil, fmt.Errorf("pulling %s's worker %s: %w", module, w.Consumer, err)
+		}
+		return nil, nil
+	}
+	if !json.Valid(msg.Data()) {
+		_ = msg.Term()
+		return nil, nil
+	}
+	headers := map[string]string{}
+	for k := range msg.Headers() {
+		headers[k] = msg.Headers().Get(k)
+	}
+	delivered := uint64(1)
+	if meta, merr := msg.Metadata(); merr == nil && meta != nil {
+		delivered = meta.NumDelivered
+	}
+	p := &Pulled{ID: pullID(), Worker: w.Consumer, Subject: msg.Subject(), Body: json.RawMessage(msg.Data()),
+		Headers: headers, Delivered: delivered}
+	seatPulls.mu.Lock()
+	seatPulls.held[p.ID] = pulledMsg{module: module, msg: msg}
+	seatPulls.mu.Unlock()
+	return p, nil
+}
+
+func (c *Conn) pullConsumer(module string, w SeatWorker) (jetstream.Consumer, error) {
+	key := module + " " + w.Consumer
+	seatPulls.mu.Lock()
+	bound, ok := seatPulls.cons[key]
+	seatPulls.mu.Unlock()
+	if ok {
+		return bound, nil
+	}
+	js, err := jetstream.New(c.nc)
+	if err != nil {
+		return nil, err
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	bound, err = js.Consumer(ctx, w.Stream, w.Consumer)
+	if err != nil {
+		return nil, fmt.Errorf("binding %s's worker %s on %s: %w", module, w.Consumer, w.Stream, err)
+	}
+	seatPulls.mu.Lock()
+	seatPulls.cons[key] = bound
+	seatPulls.mu.Unlock()
+	return bound, nil
+}
+
+// The fates SeatSettle gives a pulled message.
+const (
+	SettleAck      = "ack"      // done: never delivered again
+	SettleNak      = "nak"      // handed back, to be delivered again at once
+	SettleTerm     = "term"     // dropped
+	SettleProgress = "progress" // still being worked: its acknowledgement wait starts again
+)
+
+// SeatSettle gives the message module pulled under id its fate. A message pulled for another module, or settled
+// already, is refused.
+func (c *Conn) SeatSettle(module, id, fate string) error {
+	seatPulls.mu.Lock()
+	p, ok := seatPulls.held[id]
+	if ok && p.module != module {
+		ok = false
+	}
+	if ok && fate != SettleProgress {
+		delete(seatPulls.held, id)
+	}
+	seatPulls.mu.Unlock()
+	if !ok {
+		return fmt.Errorf("%s holds no pulled message %s", module, id)
+	}
+	switch fate {
+	case SettleAck:
+		return p.msg.Ack()
+	case SettleNak:
+		return p.msg.Nak()
+	case SettleTerm:
+		return p.msg.Term()
+	case SettleProgress:
+		return p.msg.InProgress()
+	}
+	seatPulls.mu.Lock()
+	seatPulls.held[id] = p // an unknown fate settles nothing
+	seatPulls.mu.Unlock()
+	return fmt.Errorf("a pulled message's fate is ack, nak, term or progress, not %q", fate)
+}

node-tools/internal/bus/seat_pull_test.go

@@ -0,0 +1,64 @@
+package bus_test
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/novox/mesh-tools/node-tools/internal/bus"
+)
+
+// 1887/A11: a holder pulls one message at a time, only from a worker its membership lists, never a prefetch; it settles
+// only what it pulled, and a message handed back comes again.
+func TestAHolderPullsOneMessageWhenItHasRoom(t *testing.T) {
+	_, conn, js := seatMesh(t)
+	for _, id := range []string{"a1", "a2"} {
+		if _, err := conn.SeatPublish("mesh-delivery", "mesh.seat.operator-channel.accept.ask.mesh-delivery",
+			json.RawMessage(`{"id":"`+id+`"}`), id); err != nil {
+			t.Fatal(err)
+		}
+	}
+	first, err := conn.SeatPull("messenger", "SEAT_OPERATOR_CHANNEL_worker", 0)
+	if err != nil || first == nil || string(first.Body) != `{"id":"a1"}` || first.Delivered != 1 {
+		t.Fatalf("the first pull = %+v, %v", first, err)
+	}
+	// One pulled, one still waiting on the stream: nothing was fetched ahead.
+	info, err := js.ConsumerInfo("SEAT_OPERATOR_CHANNEL", "SEAT_OPERATOR_CHANNEL_worker")
+	if err != nil || info.NumPending != 1 || info.NumAckPending != 1 {
+		t.Fatalf("after one pull: pending %d, in hand %d (%v); want 1 and 1", info.NumPending, info.NumAckPending, err)
+	}
+	second, err := conn.SeatPull("messenger", "SEAT_OPERATOR_CHANNEL_worker", 0)
+	if err != nil || second == nil || string(second.Body) != `{"id":"a2"}` {
+		t.Fatalf("the second pull = %+v, %v", second, err)
+	}
+	if none, err := conn.SeatPull("messenger", "SEAT_OPERATOR_CHANNEL_worker", 0); err != nil || none != nil {
+		t.Fatalf("an empty worker answered %+v, %v", none, err)
+	}
+	began := time.Now()
+	if none, err := conn.SeatPull("messenger", "SEAT_OPERATOR_CHANNEL_worker", 200*time.Millisecond); err != nil || none != nil || time.Since(began) < 150*time.Millisecond {
+		t.Fatalf("a waiting pull answered %+v, %v after %v", none, err, time.Since(began))
+	}
+	if err := conn.SeatSettle("telegram", first.ID, bus.SettleAck); err == nil {
+		t.Error("a module settled another's message")
+	}
+	if err := conn.SeatSettle("messenger", first.ID, bus.SettleProgress); err != nil {
+		t.Fatal(err)
+	}
+	if err := conn.SeatSettle("messenger", first.ID, bus.SettleAck); err != nil {
+		t.Fatal(err)
+	}
+	if err := conn.SeatSettle("messenger", first.ID, bus.SettleAck); err == nil {
+		t.Error("a message was settled twice")
+	}
+	if err := conn.SeatSettle("messenger", second.ID, bus.SettleNak); err != nil {
+		t.Fatal(err)
+	}
+	again, err := conn.SeatPull("messenger", "SEAT_OPERATOR_CHANNEL_worker", time.Second)
+	if err != nil || again == nil || string(again.Body) != `{"id":"a2"}` || again.Delivered != 2 {
+		t.Fatalf("the message handed back = %+v, %v", again, err)
+	}
+	if _, err := conn.SeatPull("telegram", "SEAT_OPERATOR_CHANNEL_worker", 0); err == nil || !strings.Contains(err.Error(), "membership lists") {
+		t.Errorf("a channel pulled the router's work: %v", err)
+	}
+}

node-tools/internal/launch/launch.go

@@ -288,7 +288,7 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an
 							subscriber = c
 							mu.Unlock()
 							err = mesh.Subscribe(deliver)
-						case "mesh/seat.publish", "mesh/seat.prove", "mesh/seat.record", "mesh/seat.kinds":
+						case "mesh/seat.publish", "mesh/seat.prove", "mesh/seat.record", "mesh/seat.kinds", "mesh/seat.pull", "mesh/seat.settle":
 							var answered json.RawMessage
 							if answered, err = mesh.Seat(strings.TrimPrefix(m.Method, "mesh/seat."), m.Params, nil); err == nil {
 								result = answered

node-tools/internal/launch/seat_test.go

@@ -69,3 +69,45 @@ func TestABundleTakesASeatsWorkThroughItsOwnChannel(t *testing.T) {
 		t.Fatalf("the work was not handed to the bundle: %s %v", got, err)
 	}
 }
+
+// A bundle that pulls one ask from its worker when it is listed.
+const pullingBundle = `#!/bin/sh
+while IFS= read -r line; do
+  id=$(printf '%s' "$line" | sed -n 's/.*"id":\([0-9]*\)[,}].*/\1/p')
+  case "$line" in
+    *'"method":"initialize"'*) printf '{"jsonrpc":"2.0","id":%s,"result":{}}\n' "$id" ;;
+    *'"method":"tools/list"'*)
+      printf '{"jsonrpc":"2.0","id":%s,"result":{"tools":[]}}\n' "$id"
+      printf '{"jsonrpc":"2.0","id":"p1","method":"mesh/seat.pull","params":{"worker":"SEAT_NODE_BUILD_GO_DELIVERY_worker","wait_ms":0}}\n' ;;
+  esac
+done
+`
+
+// 1887/A11: a bundle's pull reaches the runtime's bus from its own channel, as its other seat traffic does.
+func TestABundlePullsASeatsWorkThroughItsOwnChannel(t *testing.T) {
+	entry := filepath.Join(t.TempDir(), "bundle")
+	if err := os.WriteFile(entry, []byte(pullingBundle), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	b := &seating{}
+	l, err := Start("mesh-build-go", entry, os.Environ(), b, t.Logf)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(l.Stop)
+	for i := 0; ; i++ {
+		b.mu.Lock()
+		verb, params := b.verb, b.params
+		b.mu.Unlock()
+		if verb == "pull" {
+			if params != `{"worker":"SEAT_NODE_BUILD_GO_DELIVERY_worker","wait_ms":0}` {
+				t.Fatalf("pulled with %s", params)
+			}
+			return
+		}
+		if i > 100 {
+			t.Fatal("the bundle's pull never reached the runtime")
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+}

node-tools/internal/runtime/runtime.go

@@ -14,6 +14,7 @@ import (
 	"sort"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/novox/mesh-tools/node-tools/internal/announce"
 	"github.com/novox/mesh-tools/node-tools/internal/bus"
@@ -759,6 +760,9 @@ type seatAsked struct {
 	Worker  string          `json:"worker"`
 	Bucket  string          `json:"bucket"`
 	Key     string          `json:"key"`
+	// WaitMS is how long a pull waits for a message (none when 0); Fate what a settle gives one (issue 1887 A11).
+	WaitMS int    `json:"wait_ms"`
+	Fate   string `json:"fate"`
 }
 
 // Seat answers a bundle's seat traffic, each as the module and only as far as its membership lists:
@@ -792,6 +796,19 @@ func (b *moduleBus) Seat(verb string, params json.RawMessage, handOn func(string
 			return json.RawMessage("null"), err
 		}
 		return json.Marshal(entry)
+	case "pull":
+		// One message from a worker the membership lists, never a prefetch: a build module pulls an ask only when it
+		// has room, from its workers in their order (novox/hq ADR 0525 §4; issue 1887 A11).
+		p, err := conn.SeatPull(b.module, asked.Worker, time.Duration(asked.WaitMS)*time.Millisecond)
+		if err != nil || p == nil {
+			return json.RawMessage("null"), err
+		}
+		return json.Marshal(p)
+	case "settle":
+		if err := conn.SeatSettle(b.module, asked.ID, asked.Fate); err != nil {
+			return nil, err
+		}
+		return json.RawMessage(`{}`), nil
 	case "take":
 		key := "take " + asked.Worker
 		if b.bound(key) {

node-tools/internal/runtime/seat_traffic_test.go

@@ -147,3 +147,65 @@ func TestALostCompareAndSetCarriesItsCode(t *testing.T) {
 		t.Error("an update of revision zero was taken")
 	}
 }
+
+// 1887/A11: a bundle pulls one message from its own worker and settles it through its module's bus; another module may
+// neither pull that worker nor settle what it did not pull.
+func TestABundlePullsAndSettlesThroughItsModulesBus(t *testing.T) {
+	mesh := mt.New(t)
+	nc, err := nats.Connect(mt.URL(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(nc.Close)
+	js, _ := nc.JetStream()
+	if _, err := js.AddStream(&nats.StreamConfig{Name: "SEAT_NODE_BUILD_GO", Retention: nats.WorkQueuePolicy,
+		Subjects: []string{"mesh.seat.node-build-go.accept.>"}}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := js.AddConsumer("SEAT_NODE_BUILD_GO", &nats.ConsumerConfig{Durable: "SEAT_NODE_BUILD_GO_DELIVERY_worker",
+		AckPolicy: nats.AckExplicitPolicy, FilterSubject: "mesh.seat.node-build-go.accept.delivery"}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := js.Publish("mesh.seat.node-build-go.accept.delivery", []byte(`{"id":"ask1"}`)); err != nil {
+		t.Fatal(err)
+	}
+	holder := mt.MembershipOf("mesh-build-go", "anchor", false, nil)
+	holder.SeatTraffic = &bus.SeatTraffic{Workers: []bus.SeatWorker{{Stream: "SEAT_NODE_BUILD_GO",
+		Consumer: "SEAT_NODE_BUILD_GO_DELIVERY_worker", Filter: "mesh.seat.node-build-go.accept.delivery"}}}
+	other := mt.MembershipOf("mesh-build-node", "anchor", false, nil)
+	other.SeatTraffic = &bus.SeatTraffic{}
+	mesh.Issue(t, holder)
+	mesh.Issue(t, other)
+	conn := connect(t, "node-tools", "anchor")
+	conn.Follow("mesh-build-go")
+	conn.Follow("mesh-build-node")
+	mt.Until(t, func() error {
+		if conn.Membership("mesh-build-go") == nil || conn.Membership("mesh-build-node") == nil {
+			return errors.New("not issued yet")
+		}
+		return nil
+	})
+	all := &consumers{conn: conn, logf: t.Logf, of: map[string]*moduleEvents{}}
+	building := all.forModule("mesh-build-go").(*moduleBus)
+	elsewhere := all.forModule("mesh-build-node").(*moduleBus)
+	if _, err := elsewhere.Seat("pull", json.RawMessage(`{"worker":"SEAT_NODE_BUILD_GO_DELIVERY_worker"}`), nil); err == nil {
+		t.Fatal("another module pulled the Go build seat's worker")
+	}
+	raw, err := building.Seat("pull", json.RawMessage(`{"worker":"SEAT_NODE_BUILD_GO_DELIVERY_worker","wait_ms":1000}`), nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var p bus.Pulled
+	if err := json.Unmarshal(raw, &p); err != nil || string(p.Body) != `{"id":"ask1"}` || p.ID == "" {
+		t.Fatalf("pulled %s (%v)", raw, err)
+	}
+	if _, err := elsewhere.Seat("settle", json.RawMessage(`{"id":"`+p.ID+`","fate":"ack"}`), nil); err == nil {
+		t.Fatal("another module settled the pulled ask")
+	}
+	if _, err := building.Seat("settle", json.RawMessage(`{"id":"`+p.ID+`","fate":"ack"}`), nil); err != nil {
+		t.Fatal(err)
+	}
+	if none, err := building.Seat("pull", json.RawMessage(`{"worker":"SEAT_NODE_BUILD_GO_DELIVERY_worker"}`), nil); err != nil || string(none) != "null" {
+		t.Fatalf("an empty worker answered %s, %v", none, err)
+	}
+}