Merge branch 'feat/47-G-fix11-a5' into feat/47-G-join11

# Conflicts: # node-tools/internal/firstview/firstview_test.go # node-tools/internal/runtime/announce_test.go # node-tools/internal/runtime/busstate.go # node-tools/internal/runtime/ledgers_test.go
2026-11-12 10:27:27 +01:00
parent 8b980a9480 e63d707eac
commit a91386b3b5
24 changed files with 967 additions and 91 deletions
@@ -849,6 +849,12 @@ func answerOf(data []byte) (Answered, error) {
 // PublishAs emits an event as a module: published into JetStream and awaited, de-duplicated by its
 // own id (ADR 0042). The body is the payload; the metadata rides as headers.
 func (c *Conn) PublishAs(module string, env Envelope) error {
+	_, err := c.PublishAsAcked(module, env)
+	return err
+}
+
+// PublishAsAcked is PublishAs that also answers the sequence the bus acknowledged the event at.
+func (c *Conn) PublishAsAcked(module string, env Envelope) (uint64, error) {
 	// On the module's own stream when its membership names one carrying the event (ADR 0363 §2), on EVENTS
 	// otherwise — and only there, as before, for a membership that names no stream.
 	msg := nats.NewMsg(eventSubject(c.Membership(module), module, env.Key))
@@ -870,8 +876,11 @@ func (c *Conn) PublishAs(module string, env Envelope) error {
 	if id := env.Headers["x-event-id"]; id != "" {
 		opts = append(opts, nats.MsgId(id))
 	}
-	_, err := c.js.PublishMsg(msg, opts...)
-	return err
+	ack, err := c.js.PublishMsg(msg, opts...)
+	if err != nil {
+		return 0, err
+	}
+	return ack.Sequence, nil
 }

 // Say publishes one message on core NATS, kept by nobody: a heartbeat, whose loss is the next one.
@@ -10,6 +10,7 @@ import (
 	"strings"
 	"time"

+	"git.novox.be/novox/mesh-sdk/go/buscalls"
 	"git.novox.be/novox/mesh-sdk/go/buscompose"
 	"git.novox.be/novox/mesh-sdk/go/membership"
 	"github.com/nats-io/nats.go"
@@ -143,6 +144,76 @@ func (h *busHistory) Read(keys string, from uint64, limit int) ([]ledger.Kept, e
 	}
 }

+// Purge removes every operation of one instance from the history stream, on the module's ledger connection, the one
+// granted the purge of its own ledger's streams.
+func (h *busHistory) Purge(key string) error {
+	return purgeSubject(h.read, h.write, h.stream, h.prefix+"."+key)
+}
+
+// Fill is the history stream's fill against its cap, the cap the manifest gives the ledger.
+func (h *busHistory) Fill() (buscalls.Fill, error) {
+	name := h.stream
+	if name == "" {
+		ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
+		defer cancel()
+		js, err := h.read.JetStream()
+		if err != nil {
+			return buscalls.Fill{}, err
+		}
+		if name, err = js.StreamNameBySubject(ctx, h.prefix+".>"); err != nil {
+			return buscalls.Fill{}, fmt.Errorf("no stream on the bus keeps %s: %w", h.prefix, err)
+		}
+	}
+	f, err := h.read.streamFill(name)
+	return f.Fill, err
+}
+
+// purgeSubject purges one subject's messages from the stream that keeps it: the stream named, or found on the runner's
+// own connection by its subject, purged on the ledger connection.
+func purgeSubject(read, write *Conn, stream, subject string) error {
+	if write == nil {
+		return errors.New("a ledger's streams are purged only on its module's ledger connection, and none is held")
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
+	defer cancel()
+	if stream == "" {
+		js, err := read.JetStream()
+		if err != nil {
+			return err
+		}
+		if stream, err = js.StreamNameBySubject(ctx, subject); err != nil {
+			return fmt.Errorf("no stream on the bus keeps %s: %w", subject, err)
+		}
+	}
+	// The purge is asked of the stream by its name, with no read of its info first: the ledger account is granted the
+	// purge of its own ledger's two streams and nothing else (PurgeAPI).
+	body, err := json.Marshal(map[string]string{"filter": subject})
+	if err != nil {
+		return err
+	}
+	reply, err := write.nc.RequestWithContext(ctx, PurgeAPI(stream), body)
+	if err != nil {
+		return fmt.Errorf("purging %s from %s: %w", subject, stream, err)
+	}
+	var ack struct {
+		Success bool `json:"success"`
+		Error   *struct {
+			Description string `json:"description"`
+		} `json:"error"`
+	}
+	if err := json.Unmarshal(reply.Data, &ack); err != nil || !ack.Success {
+		why := string(reply.Data)
+		if ack.Error != nil {
+			why = ack.Error.Description
+		}
+		return fmt.Errorf("the bus refused the purge of %s from %s: %s", subject, stream, why)
+	}
+	return nil
+}
+
+// PurgeAPI is the bus's subject that purges a stream, the one a ledger account is granted for its own ledger's streams.
+func PurgeAPI(stream string) string { return "$JS.API.STREAM.PURGE." + stream }
+
 type busState struct {
 	c      *Conn
 	bucket string
@@ -231,6 +302,11 @@ func (s *busSnapshots) Get(key string) ([]byte, uint64, error) {
 	return lastOn(s.read, "", s.prefix+"."+key)
 }

+// Purge removes one instance's snapshot, on the module's ledger connection.
+func (s *busSnapshots) Purge(key string) error {
+	return purgeSubject(s.read, s.write, "", s.prefix+"."+key)
+}
+
 func (s *busSnapshots) Put(key string, value []byte, expect uint64) error {
 	if s.write == nil {
 		return errors.New("a ledger's snapshot is published only on its module's ledger connection, and none is held")
@@ -0,0 +1,87 @@
+package bus
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"slices"
+	"strings"
+
+	"git.novox.be/novox/mesh-sdk/go/buscompose"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// SubscribeNew hears one event, named `<emitter>.<event>`, from the next message on, on an ordered consumer of its own on each
+// stream that carries it, and hands every message to deliver as the envelope a module sees. The subjects the name lands on are
+// mesh-sdk's one rule (buscompose.EventSubjects, design 47 G §559, A-153), so a name with a wildcard is refused, naming the rule;
+// a name no stream on the bus carries is an error, never a subscription that hears nothing. stop ends every consumer.
+func (c *Conn) SubscribeNew(name string, deliver func(json.RawMessage) error) (func(), error) {
+	subjects, err := buscompose.EventSubjects(name, "")
+	if err != nil {
+		return nil, err
+	}
+	js, err := c.JetStream()
+	if err != nil {
+		return nil, err
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
+	defer cancel()
+	// The streams that carry each subject, found by the bus: a namespace no stream carries hears nothing and is skipped.
+	byStream := map[string][]string{}
+	for _, subject := range subjects {
+		if strings.HasSuffix(subject, ".>") {
+			continue
+		}
+		stream, err := js.StreamNameBySubject(ctx, subject)
+		if err != nil || stream == "" {
+			continue
+		}
+		byStream[stream] = append(byStream[stream], subject, subject+".>")
+	}
+	if len(byStream) == 0 {
+		return nil, fmt.Errorf("no stream on the bus carries %s (%s): nothing would be heard", name, strings.Join(subjects, ", "))
+	}
+	var stops []func()
+	stopAll := func() {
+		for _, s := range stops {
+			s()
+		}
+	}
+	for _, stream := range slices.Sorted(mapKeys(byStream)) {
+		cons, err := js.OrderedConsumer(ctx, stream, jetstream.OrderedConsumerConfig{FilterSubjects: byStream[stream],
+			DeliverPolicy: jetstream.DeliverNewPolicy})
+		if err != nil {
+			stopAll()
+			return nil, fmt.Errorf("an ordered consumer of %s on %s: %w", name, stream, err)
+		}
+		cc, err := cons.Consume(func(m jetstream.Msg) {
+			env, ok := envelopeOf(m.Subject(), m.Headers(), m.Data())
+			if !ok {
+				return
+			}
+			raw, err := json.Marshal(env)
+			if err != nil {
+				return
+			}
+			if err := deliver(raw); err != nil && c.Logf != nil {
+				c.Logf("[mesh-tools] a subscription for new %s messages did not take one: %v", name, err)
+			}
+		})
+		if err != nil {
+			stopAll()
+			return nil, fmt.Errorf("consuming %s on %s: %w", name, stream, err)
+		}
+		stops = append(stops, cc.Stop)
+	}
+	return stopAll, nil
+}
+
+func mapKeys(m map[string][]string) func(func(string) bool) {
+	return func(yield func(string) bool) {
+		for k := range m {
+			if !yield(k) {
+				return
+			}
+		}
+	}
+}