jschoubben
3f538027e9
Replaces the runtime's own bucket listing, advisory hearing and per-bucket bookkeeping; the bound on progress and the runtime's shapes stay here. ADR 0593 §6. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
106 lines
3.4 KiB
Go
106 lines
3.4 KiB
Go
package bus
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/pernode"
|
|
)
|
|
|
|
// A state its owner keeps one bucket per node (novox/hq ADR 0593 §6: each node's engine writes its own `composed-<node>`,
|
|
// and the provisioner, the assigner and the view read them all) is served to a bundle as one state. The rules (a key's
|
|
// first part names its node's bucket, keys and a watch over every bucket of the prefix, a bucket created later heard by
|
|
// the bus's creation advisory) are mesh-sdk's pernode package, written once for a process and for this runtime; this
|
|
// file adapts its entries to the runtime's shapes. A reader is granted the listing of stream names
|
|
// (pernode.ListSubject) and the creation advisory (pernode.CreatedAdvisory) beside each bucket's read.
|
|
|
|
func perNodeRefusal(module, name, what string) error {
|
|
return fmt.Errorf("%s's state %s is kept one bucket per node and read only: no %s of it as one state (novox/hq ADR 0593 §6)",
|
|
module, name, what)
|
|
}
|
|
|
|
func (c *Conn) family(s StateIssued) pernode.Family {
|
|
return pernode.Family{Read: s.Name, Prefix: s.Bucket, Bus: pernode.OverNATS(c.nc)}
|
|
}
|
|
|
|
func (c *Conn) perNodeGet(s StateIssued, key string) (*StateEntry, error) {
|
|
ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
|
|
defer cancel()
|
|
e, err := c.family(s).Get(ctx, key)
|
|
if err != nil || e == nil {
|
|
return nil, err
|
|
}
|
|
return &StateEntry{Key: e.Key(), Value: valueOf(e.Value()), Revision: e.Revision()}, nil
|
|
}
|
|
|
|
func (c *Conn) perNodeKeys(s StateIssued) ([]string, error) {
|
|
ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
|
|
defer cancel()
|
|
return c.family(s).Keys(ctx)
|
|
}
|
|
|
|
// perNodeWatch is StateWatch over every bucket of the family, and over a bucket created later. As in a single bucket's
|
|
// watch, StateTimeout bounds the time without progress (issue 1726), and a value the bundle does not take is said and
|
|
// the watch goes on.
|
|
func (c *Conn) perNodeWatch(module string, s StateIssued, pattern string, deliver func(StateChange) error) (func(), error) {
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
progress := make(chan struct{}, 1)
|
|
handle := func(ch pernode.Change) error {
|
|
select {
|
|
case progress <- struct{}{}:
|
|
default:
|
|
}
|
|
e := ch.Entry
|
|
change := StateChange{State: s.Name, Key: e.Key(), Revision: e.Revision(), Current: ch.Current}
|
|
switch e.Operation() {
|
|
case jetstream.KeyValuePut:
|
|
change.Op = "put"
|
|
change.Value = valueOf(e.Value())
|
|
default:
|
|
if ch.Current {
|
|
return nil
|
|
}
|
|
change.Op = "delete"
|
|
if e.Operation() == jetstream.KeyValuePurge {
|
|
change.Op = "expire"
|
|
}
|
|
}
|
|
if ctx.Err() != nil {
|
|
return nil
|
|
}
|
|
if err := deliver(change); err != nil {
|
|
c.Logf("[mesh-tools] %s did not take %s %s of %s: %v; its next change, or its next start, reads it again",
|
|
module, change.Op, change.Key, s.Name, err)
|
|
}
|
|
return nil
|
|
}
|
|
done := make(chan error, 1)
|
|
go func() { done <- c.family(s).Watch(ctx, pattern, handle) }()
|
|
bound := time.NewTimer(StateTimeout)
|
|
defer bound.Stop()
|
|
for {
|
|
select {
|
|
case err := <-done:
|
|
if err != nil {
|
|
cancel()
|
|
return nil, err
|
|
}
|
|
return cancel, nil
|
|
case <-progress:
|
|
if !bound.Stop() {
|
|
select {
|
|
case <-bound.C:
|
|
default:
|
|
}
|
|
}
|
|
bound.Reset(StateTimeout)
|
|
case <-bound.C:
|
|
cancel()
|
|
return nil, fmt.Errorf("the current values of %s did not arrive in %s: none was handed over in that time", s.Name, StateTimeout)
|
|
}
|
|
}
|
|
}
|