diff --git a/internal/link/engineseat.go b/internal/link/engineseat.go index 43605eb..045e94e 100644 --- a/internal/link/engineseat.go +++ b/internal/link/engineseat.go @@ -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 }