diff --git a/node-tools/internal/bus/bus.go b/node-tools/internal/bus/bus.go index ad508f0..55d73ce 100644 --- a/node-tools/internal/bus/bus.go +++ b/node-tools/internal/bus/bus.go @@ -366,6 +366,13 @@ func (c *Conn) Follow(module string) { // GrantsPatience bounds the first read of a membership from the grants part's state. var GrantsPatience = 2 * time.Second +// GrantsRetry is the first wait before the memberships state is read again after it could not be; it doubles up to +// GrantsRetryMost. +var ( + GrantsRetry = time.Second + GrantsRetryMost = 30 * time.Second +) + // issue applies a membership that arrived for module, live, and re-serves on it. func (c *Conn) issue(module string, m Membership) { c.mu.Lock() @@ -383,9 +390,38 @@ func (c *Conn) issue(module string, m Membership) { } // followGrants reads module's key of the grants part's state memberships and watches it: the value there now is applied -// before Follow returns, each change after it live. A state that is not there yet, or not readable, is said once, and -// the module has no membership until it can be read. +// before Follow returns, each change after it live. A state that is not there yet, or not readable (a bus answering 503 +// right after its boot, a timeout), is said once and read again, with a growing wait, until it can be or the +// connection closes: Follow runs once per module, so nothing else would ever read it again. func (c *Conn) followGrants(module string) { + if c.watchGrants(module, true) { + return + } + go func() { + wait := GrantsRetry + for !c.nc.IsClosed() { + time.Sleep(wait) + if c.nc.IsClosed() { + return + } + if c.watchGrants(module, false) { + c.Logf("[mesh-tools] the grants part's memberships are read again; %s on %s follows its key", module, c.node) + return + } + if wait *= 2; wait > GrantsRetryMost { + wait = GrantsRetryMost + } + } + }() +} + +// watchGrants is one attempt of followGrants: true once the key is watched. A failure is said when say is true. +func (c *Conn) watchGrants(module string, say bool) bool { + logf := func(format string, a ...any) { + if say { + c.Logf(format, a...) + } + } key := buscompose.MembershipKey(c.node, module) // First the key's direct get, which the account is granted alone: a bus that refuses it, or a state not there yet, // answers at once, where binding the bucket would wait out its bound on a refusal. @@ -396,27 +432,27 @@ func (c *Conn) followGrants(module string) { if err != nil { why = err.Error() } - c.Logf("[mesh-tools] the grants part's memberships (%s) cannot be read (%s); %s has no membership "+ + logf("[mesh-tools] the grants part's memberships (%s) cannot be read (%s); %s has no membership "+ "until it can be", bucket, why, module) - return + return false } js, err := jetstream.New(c.nc) if err != nil { - return + return false } ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) defer cancel() kv, err := js.KeyValue(ctx, bucket) if err != nil { - c.Logf("[mesh-tools] the grants part's memberships (%s) cannot be read (%v); %s has no membership "+ + logf("[mesh-tools] the grants part's memberships (%s) cannot be read (%v); %s has no membership "+ "until it can be", buscompose.GrantsMembershipsBucket, err, module) - return + return false } w, err := kv.Watch(context.Background(), key) if err != nil { - c.Logf("[mesh-tools] %s's key %s of the grants part's memberships cannot be watched (%v); %s has no membership "+ + logf("[mesh-tools] %s's key %s of the grants part's memberships cannot be watched (%v); %s has no membership "+ "until it can be", module, key, err, module) - return + return false } apply := func(e jetstream.KeyValueEntry) { if e.Operation() != jetstream.KeyValuePut { @@ -443,7 +479,7 @@ func (c *Conn) followGrants(module string) { select { case e, ok := <-w.Updates(): if !ok { - return + return false } if e == nil { current = false @@ -452,9 +488,9 @@ func (c *Conn) followGrants(module string) { apply(e) case <-timer.C: _ = w.Stop() - c.Logf("[mesh-tools] %s's key %s of the grants part's memberships did not answer in %s; %s has no membership "+ + logf("[mesh-tools] %s's key %s of the grants part's memberships did not answer in %s; %s has no membership "+ "until it can be", module, key, StateTimeout, module) - return + return false } } go func() { @@ -464,6 +500,7 @@ func (c *Conn) followGrants(module string) { } } }() + return true } // Membership is what the mesh issued a module here, or nil when nothing has been issued. diff --git a/node-tools/internal/bus/streams_test.go b/node-tools/internal/bus/streams_test.go index 87deff4..94a6d7c 100644 --- a/node-tools/internal/bus/streams_test.go +++ b/node-tools/internal/bus/streams_test.go @@ -199,6 +199,29 @@ func TestAModuleHearsEveryStreamItsMembershipNames(t *testing.T) { if got := consumerOf(); got != "from-grants-2" { t.Fatalf("a membership said on the controller's subject was applied: the membership names %q", got) } + // Review g1 round 1: memberships that cannot be read at the follow (a bus not ready after its boot) are read again until + // they can be; the follow, which runs once per module, does not give the module up for good. + was := bus.GrantsRetry + bus.GrantsRetry = 50 * time.Millisecond + t.Cleanup(func() { bus.GrantsRetry = was }) + mesh.DropGrants(t) + late, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"}) + if err != nil { + t.Fatal(err) + } + late.Logf = func(string, ...any) {} + t.Cleanup(late.Close) + late.Follow("reader") + if m := late.Membership("reader"); m != nil { + t.Fatalf("a membership was served from a state not there: %+v", m) + } + mesh.Issue(t, granted) + mt.Until(t, func() error { + if m := late.Membership("reader"); m == nil || len(m.Hears) != 1 || m.Hears[0].Consumer != "from-grants-2" { + return fmt.Errorf("the membership is %+v, want the one written once the state is there", m) + } + return nil + }) } // novox/hq ADR 0363 ยง6 (issue 725): a consumer the membership names before the controller made it is bound once diff --git a/node-tools/internal/meshtest/meshtest.go b/node-tools/internal/meshtest/meshtest.go index a058c77..448e27a 100644 --- a/node-tools/internal/meshtest/meshtest.go +++ b/node-tools/internal/meshtest/meshtest.go @@ -131,6 +131,14 @@ func (m *Mesh) putGrants(t *testing.T, mem bus.Membership, body []byte) { } } +// DropGrants deletes the grants part's state memberships, as a bus that has not made it yet holds none; Issue makes it again. +func (m *Mesh) DropGrants(t *testing.T) { + t.Helper() + if err := m.js.DeleteKeyValue(buscompose.GrantsMembershipsBucket); err != nil { + t.Fatal(err) + } +} + // sdkMembership is a membership of this package's type as mesh-sdk's, which the grants part composes and writes. func sdkMembership(t *testing.T, mem bus.Membership) membership.Membership { t.Helper()