H1: the node-engine publishes its seat's events again through a transport fault, under one message id (issue 1966, ADR 0668)
One publish of manifest-reads-changed with a 10 s deadline was the likely loss behind note 155; the stream takes the id once, so publishing again is safe. Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
@@ -2,6 +2,8 @@ package link
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
@@ -12,6 +14,7 @@ import (
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/busretry"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/seatlist"
|
||||
"github.com/nats-io/nats.go"
|
||||
@@ -198,15 +201,30 @@ func EngineEndpointMetadata(v catalogue.Verb, node string) map[string]string {
|
||||
"tool": v.Name, "node": node, "interchangeable": "false", "description": v.Description, "schema": string(schema)}
|
||||
}
|
||||
|
||||
// SayEngine says one of the seat's events about this node, on the events stream.
|
||||
// SayEngine says one of the seat's events about this node, on the events stream, said again through a transport fault
|
||||
// (sayOnce).
|
||||
func (l *natsLink) SayEngine(event string, body any) error {
|
||||
raw, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
_, err = l.js.Publish(EngineEventSubject(event, l.node), raw, nats.Context(ctx))
|
||||
subject := EngineEventSubject(event, l.node)
|
||||
return sayOnce(func(id string) error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
_, err := l.js.Publish(subject, raw, nats.Context(ctx), nats.MsgId(id))
|
||||
return err
|
||||
}, subject, raw)
|
||||
}
|
||||
|
||||
// sayOnce publishes one event under a message id made of its subject and its body, and publishes it again through a
|
||||
// transport fault (no responders, a deadline) within the bus's retry bound: the stream takes the id once, so a publish
|
||||
// that was stored and whose acknowledgement was lost is not stored twice (novox/hq ADR 0668 §1; issue 1966 note 155, a
|
||||
// manifest-reads-changed that one publish lost was never said, and the assigner kept a bootstrapped statement).
|
||||
func sayOnce(publish func(id string) error, subject string, raw []byte) error {
|
||||
sum := sha256.Sum256(raw)
|
||||
id := subject + "." + hex.EncodeToString(sum[:12])
|
||||
_, err := busretry.DoRepeatable("saying "+subject, func() (struct{}, error) { return struct{}{}, publish(id) }, nil)
|
||||
return err
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user