Merge branch 'feat/r4g-asks' into feat/r4g

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)
+	}
+}