diff --git a/node-tools/internal/bus/seat_pull.go b/node-tools/internal/bus/seat_pull.go new file mode 100644 index 0000000..de013ba --- /dev/null +++ b/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) +} diff --git a/node-tools/internal/bus/seat_pull_test.go b/node-tools/internal/bus/seat_pull_test.go new file mode 100644 index 0000000..0aa53e6 --- /dev/null +++ b/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) + } +} diff --git a/node-tools/internal/launch/launch.go b/node-tools/internal/launch/launch.go index 5609b6f..b4c3af8 100644 --- a/node-tools/internal/launch/launch.go +++ b/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 diff --git a/node-tools/internal/launch/seat_test.go b/node-tools/internal/launch/seat_test.go index acd45a9..1d23ab5 100644 --- a/node-tools/internal/launch/seat_test.go +++ b/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) + } +} diff --git a/node-tools/internal/runtime/runtime.go b/node-tools/internal/runtime/runtime.go index 7bd9a8d..678e73a 100644 --- a/node-tools/internal/runtime/runtime.go +++ b/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) { diff --git a/node-tools/internal/runtime/seat_traffic_test.go b/node-tools/internal/runtime/seat_traffic_test.go index 3986cc5..bd80b5d 100644 --- a/node-tools/internal/runtime/seat_traffic_test.go +++ b/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) + } +}