323 lines
11 KiB
Go
323 lines
11 KiB
Go
package process
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
)
|
|
|
|
// **A part does not stop for a bucket it can work without** (novox/hq issue 1945, met twice on 2026-10-25: the delivery
|
|
// builder exited on a refused read of the gate's source and crash-looped). A grant reaches the bus with a delivery, so a
|
|
// bucket, a watch or a consumer the bus refuses, or times out on, at start is often only not granted yet: exiting for it
|
|
// stops the part until a person's hand. A part opens what it cannot do its job at all without with EssentialBucket, which
|
|
// stops its start with an error naming the bucket and its grant, and everything else with OptionalBucket, which says what
|
|
// is not heard and goes on; a watch the bus refuses on an optional bucket is said with NotWatched. Here, beside State,
|
|
// because this is where a process opens what it keeps and reads on the bus; the parts that open buckets with their own
|
|
// JetStream (the delivery parts) take the same three.
|
|
|
|
// BucketOpener is the part of the bus a start opens buckets with: a jetstream.JetStream, or a test's fake.
|
|
type BucketOpener interface {
|
|
KeyValue(ctx context.Context, bucket string) (jetstream.KeyValue, error)
|
|
}
|
|
|
|
// EssentialBucket opens a bucket part cannot work without (it cannot `does` without it); its error is EssentialErr's.
|
|
func EssentialBucket(ctx context.Context, js BucketOpener, part, name, does string) (jetstream.KeyValue, error) {
|
|
kv, err := js.KeyValue(ctx, name)
|
|
if err != nil {
|
|
return nil, EssentialErr(part, name, does, err)
|
|
}
|
|
return kv, nil
|
|
}
|
|
|
|
// EssentialErr is the error of a bucket, consumer or stream named name that part cannot work without: it says what part
|
|
// cannot do, that it stops, and the grant to look at.
|
|
func EssentialErr(part, name, does string, err error) error {
|
|
return fmt.Errorf("%s cannot be read (%v): the %s cannot %s without it, so it stops; where the bus refused it or did not "+
|
|
"answer, this module's grant of %s (its manifest's reads, state or consumes, issued with its credential) has not "+
|
|
"reached the bus yet", name, err, part, does, name)
|
|
}
|
|
|
|
// OptionalBucket opens a bucket the part works without: any error is said with what is not heard (without), and nil
|
|
// answered.
|
|
func OptionalBucket(ctx context.Context, js BucketOpener, name, without string, logf func(string, ...any)) jetstream.KeyValue {
|
|
kv, err := js.KeyValue(ctx, name)
|
|
if err != nil {
|
|
logf("the bucket %s cannot be read (%v): %s", name, err, without)
|
|
return nil
|
|
}
|
|
return kv
|
|
}
|
|
|
|
// NotWatched says that an optional bucket opened but its watch was refused or not answered: what is not heard.
|
|
func NotWatched(logf func(string, ...any), name string, err error, without string) {
|
|
logf("the bucket %s cannot be watched (%v): %s", name, err, without)
|
|
}
|
|
|
|
// ErrNotYet is a write to an awaited bucket that is not on the bus yet: the value is kept, the latest per key, and
|
|
// written once the bucket appears.
|
|
var ErrNotYet = errors.New("the bucket is not on the bus yet: the value is kept and written when it appears")
|
|
|
|
// Subscriber is the part of a connection that hears what the bus says: a *nats.Conn, or a test's fake.
|
|
type Subscriber interface {
|
|
Subscribe(subject string, handle nats.MsgHandler) (*nats.Subscription, error)
|
|
}
|
|
|
|
// createdAdvisory is the subject the bus says a bucket's stream was created on: the server's own advisory, which a
|
|
// principal is granted for each bucket it may bind (mesh-sdk buscompose StreamCreatedAdvisory, stateCreatedHeard).
|
|
func createdAdvisory(bucket string) string { return "$JS.EVENT.ADVISORY.STREAM.CREATED.KV_" + bucket }
|
|
|
|
// bindWithin bounds one bind of an awaited bucket: one STREAM.INFO.
|
|
const bindWithin = 10 * time.Second
|
|
|
|
// AwaitedBucket is a bucket a part keeps or reads that may not be on the bus at its start, bound once it is (novox/hq
|
|
// issue 1790, met 2026-10-26: since the grants part creates a manifest's buckets after a delivery, ADR 0547, a module
|
|
// restarted by that delivery opened its new state `hearing` before it existed, got "bucket not found", and never looked
|
|
// again). It binds at AwaitBucket when it can; otherwise when the bus says the bucket's stream was created (heard on
|
|
// nc), and at each use before then (one STREAM.INFO, the cheap check), never on a timer. It says once that it waits and
|
|
// once that the bucket appeared.
|
|
type AwaitedBucket struct {
|
|
ctx context.Context
|
|
js BucketOpener
|
|
name string
|
|
without string
|
|
logf func(string, ...any)
|
|
|
|
nc Subscriber
|
|
|
|
mu sync.Mutex
|
|
kv jetstream.KeyValue
|
|
refusedSaid bool
|
|
stopStatus func()
|
|
pending map[string][]byte
|
|
order []string
|
|
ready []func(jetstream.KeyValue)
|
|
sub *nats.Subscription
|
|
}
|
|
|
|
// AwaitBucket opens the bucket name now if the bus holds it, and otherwise waits for it as AwaitedBucket says; without
|
|
// is what is not done while it is missing. nc may be nil: then only a use binds it. ctx is the part's life: a bind
|
|
// after it ends is not tried.
|
|
func AwaitBucket(ctx context.Context, js BucketOpener, nc Subscriber, name, without string, logf func(string, ...any)) *AwaitedBucket {
|
|
if logf == nil {
|
|
logf = func(string, ...any) {}
|
|
}
|
|
b := &AwaitedBucket{ctx: ctx, js: js, name: name, without: without, logf: logf, pending: map[string][]byte{}}
|
|
// Heard before the first try, so a creation between the try and the subscription is not missed.
|
|
if nc != nil {
|
|
b.nc = nc
|
|
b.hear()
|
|
b.followReconnects()
|
|
}
|
|
ctxB, cancel := context.WithTimeout(ctx, bindWithin)
|
|
kv, err := js.KeyValue(ctxB, name)
|
|
cancel()
|
|
if err == nil {
|
|
b.bound(kv, false)
|
|
return b
|
|
}
|
|
logf("the bucket %s cannot be read yet (%v): %s until it appears; it is bound when the bus says it was created, or at "+
|
|
"its next use", name, err, without)
|
|
return b
|
|
}
|
|
|
|
// KV is the bound bucket, or nil while it is not on the bus; while nil, the call tries to bind it once.
|
|
func (b *AwaitedBucket) KV() jetstream.KeyValue {
|
|
b.mu.Lock()
|
|
kv := b.kv
|
|
b.mu.Unlock()
|
|
if kv != nil {
|
|
return kv
|
|
}
|
|
return b.bind()
|
|
}
|
|
|
|
// Ready runs f with the bucket: now when it is bound, otherwise once it appears (on the goroutine that bound it).
|
|
func (b *AwaitedBucket) Ready(f func(jetstream.KeyValue)) {
|
|
b.mu.Lock()
|
|
kv := b.kv
|
|
if kv == nil {
|
|
b.ready = append(b.ready, f)
|
|
}
|
|
b.mu.Unlock()
|
|
if kv != nil {
|
|
f(kv)
|
|
}
|
|
}
|
|
|
|
// Put writes key, as jetstream.KeyValue's Put. While the bucket is missing it keeps value as key's latest and answers
|
|
// ErrNotYet; the kept values are written, in the order their keys were first kept, when it appears.
|
|
func (b *AwaitedBucket) Put(ctx context.Context, key string, value []byte) (uint64, error) {
|
|
kv := b.KV()
|
|
if kv == nil {
|
|
b.mu.Lock()
|
|
if b.kv == nil {
|
|
if _, ok := b.pending[key]; !ok {
|
|
b.order = append(b.order, key)
|
|
}
|
|
b.pending[key] = append([]byte(nil), value...)
|
|
b.mu.Unlock()
|
|
return 0, fmt.Errorf("%s: %w", b.name, ErrNotYet)
|
|
}
|
|
kv = b.kv
|
|
b.mu.Unlock()
|
|
}
|
|
return kv.Put(ctx, key, value)
|
|
}
|
|
|
|
// bind tries one STREAM.INFO of the bucket and, when it is there, binds it; it answers the bucket or nil.
|
|
func (b *AwaitedBucket) bind() jetstream.KeyValue {
|
|
if b.ctx.Err() != nil {
|
|
return nil
|
|
}
|
|
ctx, cancel := context.WithTimeout(b.ctx, bindWithin)
|
|
defer cancel()
|
|
kv, err := b.js.KeyValue(ctx, b.name)
|
|
if err != nil {
|
|
return nil
|
|
}
|
|
return b.bound(kv, true)
|
|
}
|
|
|
|
// bound keeps kv as the bucket (unless another bind did first), writes what was kept and runs what waited on it.
|
|
func (b *AwaitedBucket) bound(kv jetstream.KeyValue, appeared bool) jetstream.KeyValue {
|
|
b.mu.Lock()
|
|
if b.kv != nil {
|
|
kv = b.kv
|
|
b.mu.Unlock()
|
|
return kv
|
|
}
|
|
// The kept values are written before the bucket is handed to anyone, so no later write is overtaken by an earlier.
|
|
written, failed := 0, 0
|
|
for _, key := range b.order {
|
|
ctx, cancel := context.WithTimeout(b.ctx, bindWithin)
|
|
_, err := kv.Put(ctx, key, b.pending[key])
|
|
cancel()
|
|
if err != nil {
|
|
failed++
|
|
b.logf("the bucket %s appeared but the value kept for %s could not be written: %v", b.name, key, err)
|
|
continue
|
|
}
|
|
written++
|
|
}
|
|
b.kv, b.pending, b.order = kv, nil, nil
|
|
ready, sub, stop := b.ready, b.sub, b.stopStatus
|
|
b.ready, b.sub, b.stopStatus = nil, nil, nil
|
|
b.mu.Unlock()
|
|
if sub != nil {
|
|
_ = sub.Unsubscribe()
|
|
}
|
|
if stop != nil {
|
|
stop()
|
|
}
|
|
if appeared {
|
|
b.logf("the bucket %s appeared and is bound: %d kept value(s) written, %d not", b.name, written, failed)
|
|
}
|
|
for _, f := range ready {
|
|
f(kv)
|
|
}
|
|
return kv
|
|
}
|
|
|
|
// refusalReader is the part of a *nats.Conn that says whether the bus refused a subscription: the server's refusal comes
|
|
// back after a round trip as the connection's last error, naming the subject.
|
|
type refusalReader interface {
|
|
FlushTimeout(time.Duration) error
|
|
LastError() error
|
|
}
|
|
|
|
// statusFollower is the part of a *nats.Conn that says when it connected again, without taking its reconnect handler.
|
|
type statusFollower interface {
|
|
StatusChanged(statuses ...nats.Status) chan nats.Status
|
|
RemoveStatusListener(ch chan nats.Status)
|
|
}
|
|
|
|
// hear subscribes to the bucket's creation (again), and says once, loudly, when the bus refuses it: then only a use or
|
|
// a reconnect binds the bucket.
|
|
func (b *AwaitedBucket) hear() {
|
|
subject := createdAdvisory(b.name)
|
|
b.mu.Lock()
|
|
old, bound := b.sub, b.kv != nil
|
|
b.sub = nil
|
|
b.mu.Unlock()
|
|
if old != nil {
|
|
_ = old.Unsubscribe()
|
|
}
|
|
if bound {
|
|
return
|
|
}
|
|
checks, _ := b.nc.(refusalReader)
|
|
var before error
|
|
if checks != nil {
|
|
before = checks.LastError()
|
|
}
|
|
sub, err := b.nc.Subscribe(subject, func(*nats.Msg) { go b.bind() })
|
|
if err == nil && checks != nil && checks.FlushTimeout(bindWithin) == nil {
|
|
if after := checks.LastError(); after != nil && after != before && errors.Is(after, nats.ErrPermissionViolation) &&
|
|
strings.Contains(after.Error(), subject) {
|
|
err = after
|
|
}
|
|
}
|
|
if err != nil {
|
|
if sub != nil {
|
|
_ = sub.Unsubscribe()
|
|
}
|
|
b.mu.Lock()
|
|
said := b.refusedSaid
|
|
b.refusedSaid = true
|
|
b.mu.Unlock()
|
|
if !said {
|
|
b.logf("cannot hear the creation of %s: %v; it binds at its next use, or when the connection reconnects", b.name, err)
|
|
}
|
|
return
|
|
}
|
|
b.mu.Lock()
|
|
if b.kv != nil {
|
|
b.mu.Unlock()
|
|
if sub != nil {
|
|
_ = sub.Unsubscribe()
|
|
}
|
|
return
|
|
}
|
|
b.sub, b.refusedSaid = sub, false
|
|
b.mu.Unlock()
|
|
}
|
|
|
|
// followReconnects hears the bucket's creation again and tries a bind each time the connection connects again (a grant
|
|
// reaches the bus with a new user list; a lost subscription is made again), until the bucket is bound or the part ends.
|
|
func (b *AwaitedBucket) followReconnects() {
|
|
w, ok := b.nc.(statusFollower)
|
|
if !ok {
|
|
return
|
|
}
|
|
ch := w.StatusChanged(nats.CONNECTED)
|
|
b.mu.Lock()
|
|
if b.kv != nil {
|
|
b.mu.Unlock()
|
|
w.RemoveStatusListener(ch)
|
|
return
|
|
}
|
|
b.stopStatus = func() { w.RemoveStatusListener(ch) }
|
|
b.mu.Unlock()
|
|
go func() {
|
|
for {
|
|
select {
|
|
case <-b.ctx.Done():
|
|
w.RemoveStatusListener(ch)
|
|
return
|
|
case _, open := <-ch:
|
|
if !open {
|
|
return
|
|
}
|
|
b.hear()
|
|
b.KV()
|
|
}
|
|
}
|
|
}()
|
|
}
|