diff --git a/go/buscompose/buscompose_test.go b/go/buscompose/buscompose_test.go index e7ff21c..1d361ca 100644 --- a/go/buscompose/buscompose_test.go +++ b/go/buscompose/buscompose_test.go @@ -5,6 +5,7 @@ import ( "os" "path/filepath" "reflect" + "regexp" "strings" "testing" ) @@ -14,6 +15,9 @@ import ( // internal/broker (mesh-controller branch fix/1945-mesh-grants-publishes-memberships, and fix/1790-grants-create-buckets for the grants part creating declared streams), lists and memberships, from fixtures_test.go compiled there unchanged; composed.conf is the // controller's golden of the whole server configuration. A row that differs is the copy drifting from the controller, // which is a fault while both compose (until the controller loses its lease, ADR 0547 §13). +// createdAdvisory is one subscription to a bucket's creation in a composed list, as stateCreatedHeard composes it. +var createdAdvisory = regexp.MustCompile(`"\$JS\.EVENT\.ADVISORY\.STREAM\.CREATED\.KV_[^"]*", `) + func TestTheCopiedComposerComposesAsTheControllers(t *testing.T) { raw, err := os.ReadFile(filepath.Join("testdata", "controller-user-lists.json")) if err != nil { @@ -48,6 +52,11 @@ func TestTheCopiedComposerComposesAsTheControllers(t *testing.T) { if strings.Join(missing, ",") != strings.Join(want[i].Missing, ",") { t.Errorf("left out %v, the controller %v", missing, want[i].Missing) } + // The one row the copy composes beyond the controller's, on purpose (novox/hq issue 1790, 2026-10-26, after + // the take-over of ADR 0547 milestone 2): each principal hears its own and read buckets created, so a state + // missing at its start is bound when it appears. The controller's composer is frozen and its list no longer + // loaded; while it is not, this row is left out of the comparison and nothing else. + list = createdAdvisory.ReplaceAllString(list, "") if list != want[i].List { t.Errorf("the list differs from the controller's:\n%s", list) } diff --git a/go/buscompose/creates_streams_test.go b/go/buscompose/creates_streams_test.go index 0463e68..6ea8184 100644 --- a/go/buscompose/creates_streams_test.go +++ b/go/buscompose/creates_streams_test.go @@ -90,6 +90,22 @@ func TestTheGrantsPartCreatesOnlyDeclaredStreams(t *testing.T) { t.Errorf("%s may publish %s: %v, want %v", row.who, row.subject, got, row.may) } } + // Issue 1790 (2026-10-26): a module hears the bus create each bucket it may bind, its own and those it reads, so a + // state missing at its start is bound when it appears; no other bucket's, and no other advisory. + for _, row := range []struct { + who, subject string + may bool + }{ + {"mesh-delivery-builder", "$JS.EVENT.ADVISORY.STREAM.CREATED.KV_mesh-delivery-builder_steps", true}, + {GrantsModule, "$JS.EVENT.ADVISORY.STREAM.CREATED.KV_mesh-vault_bus-accounts", true}, + {"mesh-delivery-builder", "$JS.EVENT.ADVISORY.STREAM.CREATED.KV_mesh-grants_assertions", false}, + {"mesh-delivery-builder", "$JS.EVENT.ADVISORY.STREAM.CREATED.LOG_mesh-delivery-builder_builds", false}, + {"mesh-delivery-builder", "$JS.EVENT.ADVISORY.STREAM.DELETED.KV_mesh-delivery-builder_steps", false}, + } { + if got := MaySubscribe(perms[row.who], row.subject); got != row.may { + t.Errorf("%s may hear %s: %v, want %v", row.who, row.subject, got, row.may) + } + } // A name that is no plain token is never granted, and a module other than the grants part is never granted one. if got := CreateStreamGrants([]string{"*", ">", "KV_a.b", "KV_a_b"}); len(got) != 2 { t.Errorf("grants for wildcard names: %v", got) diff --git a/go/buscompose/nats.go b/go/buscompose/nats.go index 40be216..1ff25c8 100644 --- a/go/buscompose/nats.go +++ b/go/buscompose/nats.go @@ -1085,8 +1085,12 @@ func PermissionsFor(p Principal) (Permissions, error) { // 5. Its state, and the state of others it reads (novox/hq ADR 0201): every one read and // watched, its own written too. - pub = append(pub, stateGrants(stateAccess{Module: p.Module, Node: p.Node, Keeps: p.State, - PerMachine: p.PerMachine, Reads: p.Reads, KeyedReads: p.KeyedReads})...) + kept := stateAccess{Module: p.Module, Node: p.Node, Keeps: p.State, PerMachine: p.PerMachine, Reads: p.Reads, + KeyedReads: p.KeyedReads} + pub = append(pub, stateGrants(kept)...) + // And it hears each of those buckets created, so one not on the bus at its start is bound when it appears + // (novox/hq issue 1790; mesh-sdk process.AwaitBucket). + sub = append(sub, stateCreatedHeard(kept)...) // And its logs (novox/hq ADR 0297): appended to and read directly, its own alone. pub = append(pub, logGrants(p.Module, p.Logs)...) diff --git a/go/buscompose/process_grants.go b/go/buscompose/process_grants.go index 44d913b..99c7b1d 100644 --- a/go/buscompose/process_grants.go +++ b/go/buscompose/process_grants.go @@ -30,7 +30,11 @@ func processGrants(p Principal) (pub, sub []string, err error) { // 3. Its module's own state, by the local names its bus names alone — never the whole of what the // module keeps (novox/hq ADR 0436 §1), read and written as stateGrants gives any module's own bucket. - pub = append(pub, stateGrants(stateAccess{Module: p.Module, Node: p.Node, Keeps: p.State})...) + kept := stateAccess{Module: p.Module, Node: p.Node, Keeps: p.State} + pub = append(pub, stateGrants(kept)...) + // And it hears each of them created, so one not on the bus at its start is bound when it appears (novox/hq issue + // 1790; process.AwaitBucket). + sub = append(sub, stateCreatedHeard(kept)...) // 4. The router's records of its module's own asks (novox/hq ADR 0436 §1, ADR 0259 §3): the same // direct-get subjects a module's own principal reads for the seats its Uses names that keep them — diff --git a/go/buscompose/state.go b/go/buscompose/state.go index 9d8bdac..f772cae 100644 --- a/go/buscompose/state.go +++ b/go/buscompose/state.go @@ -166,6 +166,49 @@ func stateGrants(a stateAccess) []string { return out } +// StreamCreatedAdvisory is the subject the bus says a stream was created on (the server's own advisory), here for a +// bucket's stream KV_. +func StreamCreatedAdvisory(bucket string) string { + return "$JS.EVENT.ADVISORY.STREAM.CREATED.KV_" + bucket +} + +// stateCreatedHeard is what a principal subscribes to so that a bucket it may bind and that is not on the bus yet is +// bound when it appears (novox/hq issue 1790, 2026-10-26: since the grants part creates a manifest's buckets, ADR 0547, +// a module's delivery can start it before its new state exists): the bus's advisory that the bucket's stream was +// created, for each bucket stateGrants lets it bind, and no other. Read-only: the server publishes it in the mesh's own +// account, and hearing it changes nothing. +func stateCreatedHeard(a stateAccess) []string { + var out []string + seen := map[string]bool{} + add := func(bucket string) { + if !seen[bucket] { + seen[bucket] = true + out = append(out, StreamCreatedAdvisory(bucket)) + } + } + perMachine := map[string]bool{} + for _, n := range a.PerMachine { + perMachine[n] = true + } + for _, name := range a.Keeps { + if !safeSubject.MatchString(name) || (perMachine[name] && !safeSubject.MatchString(a.Node)) { + continue + } + add(BucketName(a.Module, name)) + } + for _, r := range a.Reads { + if bucket, ok := bucketOfRead(r); ok { + add(bucket) + } + } + for _, k := range a.KeyedReads { + if bucket, ok := bucketOfRead(k.Read); ok && safeSubject.MatchString(k.Key) { + add(bucket) + } + } + return out +} + // KeyedRead is one key of another module's state a principal reads, and no other key of it. type KeyedRead struct { Read string `json:"read"` diff --git a/go/delivery/buildrun/lostask.go b/go/delivery/buildrun/lostask.go index 90a9ece..750acaa 100644 --- a/go/delivery/buildrun/lostask.go +++ b/go/delivery/buildrun/lostask.go @@ -3,6 +3,7 @@ package buildrun import ( "context" "encoding/json" + "errors" "fmt" "strconv" "strings" @@ -13,6 +14,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/health" + "git.novox.be/novox/mesh-sdk/go/process" ) // **An ask the build seat gave up on ends its step** (novox/hq issue 734). A build seat's worker offers an ask @@ -458,9 +460,15 @@ func (h *DeadLetterHearing) tell() { // health tool reads. const HearingState = "hearing" +// HearingKeeper is where a Hearing is kept: a jetstream.KeyValue, or a process.AwaitedBucket for a state that may not be +// on the bus at the part's start (issue 1790), which keeps the latest until it appears and answers process.ErrNotYet. +type HearingKeeper interface { + Put(ctx context.Context, key string, value []byte) (uint64, error) +} + // SayHearing is a DeadLetterHearing's said that keeps each Hearing in kv under HearingKey; nil kv keeps nothing and -// says so once. -func SayHearing(ctx context.Context, kv jetstream.KeyValue, logf func(string, ...any)) func(Hearing) { +// says so once. A write kv keeps for later (process.ErrNotYet) is not an error: the awaited bucket said it waits. +func SayHearing(ctx context.Context, kv HearingKeeper, logf func(string, ...any)) func(Hearing) { if kv == nil { if logf != nil { logf("whether %s is heard is not kept: the state %s is not there, so the health tool cannot say it", DeadLettersStream, HearingState) @@ -472,7 +480,7 @@ func SayHearing(ctx context.Context, kv jetstream.KeyValue, logf func(string, .. if err == nil { _, err = kv.Put(ctx, HearingKey, data) } - if err != nil && logf != nil { + if err != nil && !errors.Is(err, process.ErrNotYet) && logf != nil { logf("whether %s is heard could not be kept: %v", DeadLettersStream, err) } } diff --git a/go/process/account_test.go b/go/process/account_test.go index c5f9467..8b60dd8 100644 --- a/go/process/account_test.go +++ b/go/process/account_test.go @@ -8,8 +8,11 @@ import ( "os" "path/filepath" "strings" + "sync" "testing" + "time" + "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" ) @@ -73,6 +76,50 @@ func TestReadAccountRefusesAFileMissingAField(t *testing.T) { } }) + t.Run("1790: a bucket missing at start is bound when the bus says it was created, and what was written meanwhile is kept", func(t *testing.T) { + bus := &laterBus{} + hear := &heldSubscriber{} + var said []string + logf := func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) } + b := AwaitBucket(context.Background(), bus, hear, "m_hearing", "the health tool cannot say it", logf) + if hear.subject != "$JS.EVENT.ADVISORY.STREAM.CREATED.KV_m_hearing" { + t.Fatalf("heard %q", hear.subject) + } + ready := make(chan jetstream.KeyValue, 1) + b.Ready(func(kv jetstream.KeyValue) { ready <- kv }) + for _, v := range []string{"one", "two"} { + if _, err := b.Put(context.Background(), "dead-letters", []byte(v)); !errors.Is(err, ErrNotYet) { + t.Fatalf("a write while missing: %v", err) + } + } + if b.KV() != nil || len(ready) > 0 { + t.Fatal("bound before the bucket exists") + } + bus.create() + hear.handle(&nats.Msg{Subject: hear.subject}) + select { + case <-ready: + case <-time.After(5 * time.Second): + t.Fatal("not bound within 5s of the bus saying the bucket was created") + } + if got := bus.kv.puts; strings.Join(got, ",") != "dead-letters=two" { + t.Errorf("written when it appeared: %v, want the latest value only", got) + } + if _, err := b.Put(context.Background(), "dead-letters", []byte("three")); err != nil { + t.Errorf("a write once bound: %v", err) + } + if len(said) != 2 || !strings.Contains(said[0], "cannot be read yet") || !strings.Contains(said[1], "appeared") { + t.Errorf("said %q", said) + } + // Without the advisory heard, the next use binds it. + quiet := &laterBus{} + c := AwaitBucket(context.Background(), quiet, nil, "m_hearing", "x", nil) + quiet.create() + if c.KV() == nil { + t.Error("a use after the bucket appeared did not bind it") + } + }) + for _, tc := range []struct { field string values map[string]any @@ -114,3 +161,42 @@ func writeAccount(t *testing.T, fields map[string]any) string { type refusingBus struct{ err error } func (r refusingBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) { return nil, r.err } + +// laterBus has no bucket until create, and then one. +type laterBus struct { + mu sync.Mutex + kv *keptKV +} + +func (l *laterBus) create() { l.mu.Lock(); l.kv = &keptKV{}; l.mu.Unlock() } + +func (l *laterBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) { + l.mu.Lock() + defer l.mu.Unlock() + if l.kv == nil { + return nil, jetstream.ErrBucketNotFound + } + return l.kv, nil +} + +// keptKV records each put as key=value. +type keptKV struct { + jetstream.KeyValue + puts []string +} + +func (k *keptKV) Put(_ context.Context, key string, value []byte) (uint64, error) { + k.puts = append(k.puts, key+"="+string(value)) + return uint64(len(k.puts)), nil +} + +// heldSubscriber keeps the one subscription made, for the test to deliver to. +type heldSubscriber struct { + subject string + handle nats.MsgHandler +} + +func (h *heldSubscriber) Subscribe(subject string, handle nats.MsgHandler) (*nats.Subscription, error) { + h.subject, h.handle = subject, handle + return nil, nil +} diff --git a/go/process/buckets.go b/go/process/buckets.go index 24651a4..c706ba1 100644 --- a/go/process/buckets.go +++ b/go/process/buckets.go @@ -2,8 +2,12 @@ package process import ( "context" + "errors" "fmt" + "sync" + "time" + "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" ) @@ -53,3 +57,164 @@ func OptionalBucket(ctx context.Context, js BucketOpener, name, without string, 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) + + mu sync.Mutex + kv jetstream.KeyValue + 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 { + sub, err := nc.Subscribe(createdAdvisory(name), func(*nats.Msg) { go b.bind() }) + if err == nil { + b.mu.Lock() + b.sub = sub + b.mu.Unlock() + } + } + 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 := b.ready, b.sub + b.ready, b.sub = nil, nil + b.mu.Unlock() + if sub != nil { + _ = sub.Unsubscribe() + } + 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 +} diff --git a/go/process/fixture_integration_test.go b/go/process/fixture_integration_test.go index 73991c5..59ce1c2 100644 --- a/go/process/fixture_integration_test.go +++ b/go/process/fixture_integration_test.go @@ -58,7 +58,9 @@ func grantedSendProcess(t *testing.T) (url string, account Account) { {Username: adminUser, Password: adminPass}, {Username: sendUser, Password: sendPassword, Permissions: &server.Permissions{ Publish: &server.SubjectPermission{Allow: sendGrants}, - Subscribe: &server.SubjectPermission{Allow: []string{"_INBOX." + sendUser + ".>"}}, + // And the creation of its bucket, as mesh-sdk's composer grants it since issue 1790 (stateCreatedHeard). + Subscribe: &server.SubjectPermission{Allow: []string{"_INBOX." + sendUser + ".>", + "$JS.EVENT.ADVISORY.STREAM.CREATED.KV_mesh-mail_sends"}}, }}, }, } diff --git a/go/process/state_integration_test.go b/go/process/state_integration_test.go index 32c1e72..bd34388 100644 --- a/go/process/state_integration_test.go +++ b/go/process/state_integration_test.go @@ -18,12 +18,40 @@ import ( // key, reads it, and updates it with the revision it read, then each succeeds and returns the new // revision; an update with a stale revision fails as stdio.ErrChanged; a create of an existing key fails // as stdio.ErrChanged. +// +// 1790: and before that, the bucket is not on the bus when the process opens it (a delivery restarted it before the +// grants part created its new state): it is bound when the bus says the bucket was created, as the process's own +// account hears it, and the value written while it was missing is there. func TestStateCreateGetUpdateOnAGrantedBucket(t *testing.T) { url, account := grantedSendProcess(t) _, adminJetStream := adminJS(t, url) + + nc, err := nats.Connect(url, nats.UserInfo(sendUser, sendPassword), nats.CustomInboxPrefix("_INBOX."+sendUser)) + if err != nil { + t.Fatal(err) + } + defer nc.Close() + js, err := jetstream.New(nc) + if err != nil { + t.Fatal(err) + } + awaited := AwaitBucket(ctxFor(t), js, nc, "mesh-mail_sends", "nothing is kept", t.Logf) + appeared := make(chan struct{}) + awaited.Ready(func(jetstream.KeyValue) { close(appeared) }) + if _, err := awaited.Put(ctxFor(t), "hearing", []byte(`{"heard":false}`)); !errors.Is(err, ErrNotYet) { + t.Fatalf("a write before the bucket exists: %v, want ErrNotYet", err) + } + if _, err := awaited.Put(ctxFor(t), "hearing", []byte(`{"heard":true}`)); !errors.Is(err, ErrNotYet) { + t.Fatalf("a second write before the bucket exists: %v, want ErrNotYet", err) + } if _, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_sends"}); err != nil { t.Fatalf("the admin could not make the bucket: %v", err) } + select { + case <-appeared: + case <-time.After(5 * time.Second): + t.Fatal("the bucket was not bound within 5s of its creation, with no use since") + } conn, err := Connect(account) if err != nil { @@ -31,6 +59,9 @@ func TestStateCreateGetUpdateOnAGrantedBucket(t *testing.T) { } defer conn.Close() s := conn.State("sends") + if kept, err := s.Get("hearing"); err != nil || kept == nil || string(kept.Value) != `{"heard":true}` { + t.Fatalf("the value written while the bucket was missing: %+v, %v; want the latest", kept, err) + } rev, err := s.Create("key", map[string]any{"v": 1}) if err != nil {