jschoubben
45e2b946c7
The node follows the node-engine seat's running events (an ordered read of the last message of each subject, then a durable consumer), keeps what each node runs by `at`, acts on none until the first read ended, and hands the composer each node's heard build with the manifest read by the event's digest; compose resolves the passes through catalogue.SeatsToPassOn (heard, entries) with the trunk's entry for a node that said nothing. Builds: design 47 G §199, §249, §459, §536, §556 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
96 lines
3.6 KiB
Go
96 lines
3.6 KiB
Go
package link
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
|
|
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
|
"github.com/nats-io/nats.go"
|
|
)
|
|
|
|
// The node seat's event running, said by every node-engine for each module it saw running healthy, is kept on the stream
|
|
// SEAT_node-engine_running, the last message of each subject with no age limit (design 47 G §199, §260). The node reads it
|
|
// at each start through an ordered consumer it makes for that one read (§556), then follows a durable consumer of its own
|
|
// from its acknowledged position: what the other nodes run is known from the last message of each subject however long ago it
|
|
// was said, never from a have key or a call.
|
|
|
|
// RunningBus is a link that hears the seat's event running of every node.
|
|
type RunningBus interface {
|
|
// FirstRunning calls first with the last message of every subject, through an ordered consumer made for this read, and
|
|
// answers when it has read them all.
|
|
FirstRunning(ctx context.Context, first func([]byte)) error
|
|
// HearRunning calls heard with each message after the durable consumer's acknowledged position (a new consumer begins at
|
|
// the last message of each subject), until stop is called or the link ends.
|
|
HearRunning(consumer string, heard func([]byte)) (stop func(), err error)
|
|
}
|
|
|
|
// RunningConsumer is the name of this node's durable consumer of the stream.
|
|
func RunningConsumer(node string) string { return "node-engine-" + node + "-running" }
|
|
|
|
func (l *natsLink) FirstRunning(ctx context.Context, first func([]byte)) error {
|
|
sub, err := l.js.SubscribeSync(nodeengine.SubjectRunningAll, nats.OrderedConsumer(), nats.DeliverLastPerSubject(),
|
|
nats.BindStream(nodeengine.StreamRunning))
|
|
if err != nil {
|
|
return fmt.Errorf("the stream %s cannot be read: %w", nodeengine.StreamRunning, err)
|
|
}
|
|
defer sub.Unsubscribe()
|
|
info, err := sub.ConsumerInfo()
|
|
if err != nil {
|
|
return fmt.Errorf("the stream %s cannot be read: %w", nodeengine.StreamRunning, err)
|
|
}
|
|
for total, got := info.Delivered.Consumer+info.NumPending, uint64(0); got < total; got++ {
|
|
msg, err := sub.NextMsgWithContext(ctx)
|
|
if err != nil {
|
|
return fmt.Errorf("reading the last running of each subject (%d of %d): %w", got, total, err)
|
|
}
|
|
first(msg.Data)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (l *natsLink) HearRunning(consumer string, heard func([]byte)) (func(), error) {
|
|
sub, err := l.js.Subscribe(nodeengine.SubjectRunningAll, func(m *nats.Msg) {
|
|
heard(m.Data)
|
|
_ = m.Ack()
|
|
}, nats.Durable(consumer), nats.BindStream(nodeengine.StreamRunning), nats.DeliverLastPerSubject(), nats.AckExplicit(),
|
|
nats.ManualAck())
|
|
if err != nil {
|
|
return nil, fmt.Errorf("the stream %s cannot be followed: %w", nodeengine.StreamRunning, err)
|
|
}
|
|
return func() { _ = sub.Unsubscribe() }, nil
|
|
}
|
|
|
|
var _ RunningBus = (*natsLink)(nil)
|
|
|
|
// FollowRunning hands every node's running event to heard: the last of each subject first, with readEnded called after them
|
|
// (until then the caller acts on none, §556), then each after them, until ctx ends. An error is the stream or the link not
|
|
// being there, which the caller tries again.
|
|
func (q *Session) FollowRunning(ctx context.Context, heard func(nodeengine.Running), readEnded func()) error {
|
|
q.init()
|
|
q.mu.Lock()
|
|
bus := q.bus
|
|
q.mu.Unlock()
|
|
h, ok := bus.(RunningBus)
|
|
if bus == nil || !ok {
|
|
return ErrNoLink
|
|
}
|
|
decode := func(raw []byte) {
|
|
var ev nodeengine.Running
|
|
if json.Unmarshal(raw, &ev) == nil && ev.Node != "" && ev.Module != "" {
|
|
heard(ev)
|
|
}
|
|
}
|
|
if err := h.FirstRunning(ctx, decode); err != nil {
|
|
return err
|
|
}
|
|
stop, err := h.HearRunning(RunningConsumer(q.Membership.Node), decode)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer stop()
|
|
readEnded()
|
|
<-ctx.Done()
|
|
return nil
|
|
}
|