From 9d9e3a46c4f5943cfffb413d962ce48b75e2fd39 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 18 Oct 2026 19:18:08 +0200 Subject: [PATCH] Bind the have bucket once per pass, bound by the queue's timeout, and stop the pass at the first error (issue 1132) --- internal/link/bus.go | 52 ++++++++++++++++------------ internal/link/have.go | 55 +++++++++++++++++++++--------- internal/link/have_test.go | 64 +++++++++++++++++++++++++++++++---- internal/link/hearing_nats.go | 8 ++--- 4 files changed, 129 insertions(+), 50 deletions(-) diff --git a/internal/link/bus.go b/internal/link/bus.go index 7d44b8e..c07d8fd 100644 --- a/internal/link/bus.go +++ b/internal/link/bus.go @@ -118,24 +118,38 @@ func (b OverNATS) Alive(ctx context.Context, node string, body []byte) error { return b.Conn.FlushWithContext(flush) } -// haveBucket binds to the have bucket the controller made (novox/hq ADR 0429 §4, issue 1132). **Only binds**: a node-engine's -// account reaches no part of the JetStream API beyond what it is granted, and the bucket is the controller's to create -// and to configure, so a bucket that is not there is an error said as what it is. -func (b OverNATS) haveBucket() (nats.KeyValue, error) { - kv, err := b.JS.KeyValue(delivery.HaveBucket) +// HaveBind binds the have bucket the controller made (novox/hq ADR 0429 §4, issue 1132), once for a pass of writes. +// **Only binds**: a node-engine's account reaches no part of the JetStream API beyond what it is granted, and the bucket +// is the controller's to create and to configure, so a bucket that is not there is an error said as what it is. Every +// call on the bucket waits no longer than ctx has left: a refused publish is answered by nothing, and the client's own +// wait is five seconds. +func (b OverNATS) HaveBind(ctx context.Context) (HaveBucket, error) { + js := b.JS + wait := 5 * time.Second + if at, ok := ctx.Deadline(); ok { + wait = time.Until(at) + if wait <= 0 { + return nil, context.DeadlineExceeded + } + } + if b.Conn != nil { + bounded, err := b.Conn.JetStream(nats.MaxWait(wait)) + if err != nil { + return nil, err + } + js = bounded + } + kv, err := js.KeyValue(delivery.HaveBucket) if err != nil { return nil, fmt.Errorf("the have bucket %s: %w", delivery.HaveBucket, err) } - return kv, nil + return natsHaveBucket{kv}, nil } -// HaveGet reads a key of the have bucket with its revision. -func (b OverNATS) HaveGet(_ context.Context, key string) ([]byte, uint64, error) { - kv, err := b.haveBucket() - if err != nil { - return nil, 0, err - } - entry, err := kv.Get(key) +type natsHaveBucket struct{ kv nats.KeyValue } + +func (n natsHaveBucket) Get(_ context.Context, key string) ([]byte, uint64, error) { + entry, err := n.kv.Get(key) if errors.Is(err, nats.ErrKeyNotFound) { return nil, 0, ErrHaveAbsent } @@ -145,17 +159,13 @@ func (b OverNATS) HaveGet(_ context.Context, key string) ([]byte, uint64, error) return entry.Value(), entry.Revision(), nil } -// HaveSet writes a key of the have bucket over the revision read, or creates it when the revision is zero. -func (b OverNATS) HaveSet(_ context.Context, key string, value []byte, revision uint64) error { - kv, err := b.haveBucket() - if err != nil { - return err - } +func (n natsHaveBucket) Set(_ context.Context, key string, value []byte, revision uint64) error { + var err error if revision == 0 { - _, err = kv.Create(key, value) + _, err = n.kv.Create(key, value) return err } - _, err = kv.Update(key, value, revision) + _, err = n.kv.Update(key, value, revision) return err } diff --git a/internal/link/have.go b/internal/link/have.go index 365d189..99ef52b 100644 --- a/internal/link/have.go +++ b/internal/link/have.go @@ -14,13 +14,20 @@ import ( // ErrHaveAbsent is what HaveGet returns for a key the bucket does not hold. var ErrHaveAbsent = errors.New("link: no such key") -// HaveBus is a link that can read and write the node-engine's own keys in the have bucket (novox/hq ADR 0429 §4). Separate -// from Bus, as HealthBus is: a link that cannot is still a link, and the write waits for the next apply or look. +// HaveBus is a link that can bind the have bucket (novox/hq ADR 0429 §4). Separate from Bus, as HealthBus is: a link that +// cannot is still a link, and the write waits for the next apply or look. type HaveBus interface { - // HaveGet is a key's value and last revision, or ErrHaveAbsent. - HaveGet(ctx context.Context, key string) (value []byte, revision uint64, err error) - // HaveSet writes a key over the revision read: zero creates it, and a key that moved since is refused. - HaveSet(ctx context.Context, key string, value []byte, revision uint64) error + // HaveBind binds the bucket the controller made, once for a pass of writes, within ctx. A node whose grant is not there + // yet gets an error here (or the context's end), and writes nothing. + HaveBind(ctx context.Context) (HaveBucket, error) +} + +// HaveBucket is the bound bucket. +type HaveBucket interface { + // Get is a key's value and last revision, or ErrHaveAbsent. + Get(ctx context.Context, key string) (value []byte, revision uint64, err error) + // Set writes a key over the revision read: zero creates it, and a key that moved since is refused. + Set(ctx context.Context, key string, value []byte, revision uint64) error } // haveState is what the node-engine last learned of each module of the declaration it applied, and of each module's health @@ -151,8 +158,15 @@ func (q *Queue) writeHaves(ctx context.Context, bus Bus) { return } node := q.Membership.Node + // The whole pass is bounded by the queue's timeout, and it stops at the first error: a node whose grant is not there + // yet would otherwise make the worker wait the bound again for every key (issue 1132). write, cancel := context.WithTimeout(context.WithoutCancel(ctx), q.timeout()) defer cancel() + bucket, err := hb.HaveBind(write) + if err != nil { + q.sayHave(fmt.Sprintf("could not bind the have bucket, and wrote nothing there; the next apply or look writes again: %v", err)) + return + } modules := make([]string, 0, len(s.modules)) for module := range s.modules { modules = append(modules, module) @@ -178,10 +192,12 @@ func (q *Queue) writeHaves(ctx context.Context, bus Bus) { q.sayHave(fmt.Sprintf("cannot encode %s: %v", key, err)) continue } - q.putHave(write, hb, key, raw, func(current []byte) bool { + if !q.putHave(write, bucket, key, raw, func(current []byte) bool { c, err := delivery.DecodeHave(current) return err == nil && delivery.SameHave(c, h) - }) + }) { + return + } } if s.node != nil { key, err := delivery.HaveNodeKey(node) @@ -189,7 +205,7 @@ func (q *Queue) writeHaves(ctx context.Context, bus Bus) { var raw []byte if raw, err = delivery.EncodeHaveNode(*s.node); err == nil { want := *s.node - q.putHave(write, hb, key, raw, func(current []byte) bool { + q.putHave(write, bucket, key, raw, func(current []byte) bool { c, err := delivery.DecodeHaveNode(current) return err == nil && delivery.SameHaveNode(c, want) }) @@ -201,19 +217,26 @@ func (q *Queue) writeHaves(ctx context.Context, bus Bus) { } } -// putHave reads a key and writes it unless same says it already holds the value. -func (q *Queue) putHave(ctx context.Context, hb HaveBus, key string, value []byte, same func([]byte) bool) { - current, revision, err := hb.HaveGet(ctx, key) +// putHave reads a key and writes it unless same says it already holds the value. False when the pass should stop: the +// bucket failed, the error is said once, and the rest wait for the next pass. +func (q *Queue) putHave(ctx context.Context, b HaveBucket, key string, value []byte, same func([]byte) bool) bool { + if ctx.Err() != nil { + q.sayHave(fmt.Sprintf("the time for writing the have bucket ran out before %s; the next apply or look writes it", key)) + return false + } + current, revision, err := b.Get(ctx, key) switch { case errors.Is(err, ErrHaveAbsent): revision = 0 case err != nil: - q.sayHave(fmt.Sprintf("could not read %s in the have bucket, and wrote nothing there: %v", key, err)) - return + q.sayHave(fmt.Sprintf("could not read %s in the have bucket, and wrote nothing more there: %v", key, err)) + return false case same(current): - return + return true } - if err := hb.HaveSet(ctx, key, value, revision); err != nil { + if err := b.Set(ctx, key, value, revision); err != nil { q.sayHave(fmt.Sprintf("could not write %s in the have bucket; the next apply or look writes it again: %v", key, err)) + return false } + return true } diff --git a/internal/link/have_test.go b/internal/link/have_test.go index 96c8fa0..90f465b 100644 --- a/internal/link/have_test.go +++ b/internal/link/have_test.go @@ -18,11 +18,14 @@ import ( // haveLink is a link with a bucket behind it that keeps each key's revision as the bus does. type haveLink struct { *quietLink - hmu sync.Mutex - kv map[string]haveEntry - sets []string // keys written, in order - events []string // "report" and "have ", in the order they happened - failing bool + hmu sync.Mutex + kv map[string]haveEntry + sets []string // keys written, in order + events []string // "report" and "have ", in the order they happened + failing bool + binds int + setTries int + bindFails, bindStuck bool } type haveEntry struct { @@ -44,7 +47,22 @@ func (l *haveLink) Report(ctx context.Context, node string, body []byte) error { func (l *haveLink) Health(context.Context, string, []byte) error { return nil } -func (l *haveLink) HaveGet(_ context.Context, key string) ([]byte, uint64, error) { +func (l *haveLink) HaveBind(ctx context.Context) (HaveBucket, error) { + l.hmu.Lock() + l.binds++ + stuck, refuse := l.bindStuck, l.bindFails + l.hmu.Unlock() + if stuck { + <-ctx.Done() + return nil, ctx.Err() + } + if refuse { + return nil, errors.New("no grant for the have bucket") + } + return l, nil +} + +func (l *haveLink) Get(_ context.Context, key string) ([]byte, uint64, error) { l.hmu.Lock() defer l.hmu.Unlock() e, ok := l.kv[key] @@ -54,9 +72,10 @@ func (l *haveLink) HaveGet(_ context.Context, key string) ([]byte, uint64, error return e.value, e.rev, nil } -func (l *haveLink) HaveSet(_ context.Context, key string, value []byte, revision uint64) error { +func (l *haveLink) Set(_ context.Context, key string, value []byte, revision uint64) error { l.hmu.Lock() defer l.hmu.Unlock() + l.setTries++ if l.failing { return errors.New("the bucket refused") } @@ -289,3 +308,34 @@ func TestASilentReconcileStillWritesTheHaves(t *testing.T) { t.Errorf("an unchanged reconcile wrote %d keys", l.setCount()-before) } } + +// 1132/AC7: a bind that fails, or never answers, writes nothing and returns within the queue's timeout; a write that +// fails ends the pass, and is said once. +func TestAFailedBindOrWriteEndsThePassAndReturnsInTime(t *testing.T) { + for name, set := range map[string]func(*haveLink){ + "a bind refused": func(l *haveLink) { l.bindFails = true }, + "a bind that never answers": func(l *haveLink) { l.bindStuck = true }, + } { + l := newHaveLink() + set(l) + q, said := aHaveQueue(l) + q.Timeout = 100 * time.Millisecond + start := time.Now() + if !q.tell(context.Background(), appliedModules(), applied) { + t.Fatalf("%s: the report did not go out", name) + } + if took := time.Since(start); took > time.Second { + t.Errorf("%s: the pass took %s", name, took) + } + if l.binds != 1 || l.setTries != 0 || len(*said) != 1 { + t.Errorf("%s: binds %d, writes tried %d, said %v; want one bind, no write, one line", name, l.binds, l.setTries, *said) + } + } + l := newHaveLink() + l.failing = true + q, said := aHaveQueue(l) + q.tell(context.Background(), appliedModules(), applied) + if l.binds != 1 || l.setTries != 1 || len(*said) != 1 { + t.Errorf("a failed write: binds %d, writes tried %d, said %v; want one of each and one line", l.binds, l.setTries, *said) + } +} diff --git a/internal/link/hearing_nats.go b/internal/link/hearing_nats.go index ab8979d..dbe3d79 100644 --- a/internal/link/hearing_nats.go +++ b/internal/link/hearing_nats.go @@ -342,12 +342,8 @@ func (l *natsLink) Health(ctx context.Context, node string, body []byte) error { return OverNATS{Conn: l.conn, JS: l.js}.Health(ctx, node, body) } -func (l *natsLink) HaveGet(ctx context.Context, key string) ([]byte, uint64, error) { - return OverNATS{Conn: l.conn, JS: l.js}.HaveGet(ctx, key) -} - -func (l *natsLink) HaveSet(ctx context.Context, key string, value []byte, revision uint64) error { - return OverNATS{Conn: l.conn, JS: l.js}.HaveSet(ctx, key, value, revision) +func (l *natsLink) HaveBind(ctx context.Context) (HaveBucket, error) { + return OverNATS{Conn: l.conn, JS: l.js}.HaveBind(ctx) } // Ask asks a module's tool for a declared tool check (novox/hq ADR 0240, to-be 48 §3). The link the