From de320684f22891b8f9b77ded05ef670b0d5491dc Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 00:21:49 +0100 Subject: [PATCH] Grants purges the old sealed lists' keys instead of deleting them (issue 1966) A key-value delete only adds a marker: every value of the per-user keys of the lists older than the two kept stayed in the state user-list's stream, which reached 94% of its 64 MiB cap on 2026-11-08 (5,968 keys, 13,228 values). At the cap the bus refuses grants' next write of the sealed list and deliveries lock. The part now takes those keys away through a purger on the process's own JetStream connection, a publish to the key's own subject the account already writes on; the tools, with no purger, still delete. Claude-Session: https://claude.ai/code/session_01NcbgfFfQ4Z9UE5jK32PmE5 --- .../mesh-grants/composed_integration_test.go | 36 +++++++- modules/mesh-grants/cmd/mesh-grants/main.go | 2 +- modules/mesh-grants/cmd/mesh-grants/purge.go | 86 +++++++++++++++++++ modules/mesh-grants/grants_test.go | 62 +++++++++++++ modules/mesh-grants/internal/grants/grants.go | 6 +- modules/mesh-grants/internal/grants/part.go | 3 + modules/mesh-grants/internal/grants/purge.go | 23 +++++ 7 files changed, 213 insertions(+), 5 deletions(-) create mode 100644 modules/mesh-grants/cmd/mesh-grants/purge.go create mode 100644 modules/mesh-grants/internal/grants/purge.go diff --git a/modules/mesh-grants/cmd/mesh-grants/composed_integration_test.go b/modules/mesh-grants/cmd/mesh-grants/composed_integration_test.go index 69cffdca..188a48e3 100644 --- a/modules/mesh-grants/cmd/mesh-grants/composed_integration_test.go +++ b/modules/mesh-grants/cmd/mesh-grants/composed_integration_test.go @@ -7,6 +7,7 @@ import ( "context" "encoding/base64" "encoding/json" + "errors" "os" "path/filepath" "slices" @@ -100,7 +101,7 @@ func theBusNodeFollowsTheSealedListUnderTheComposedGrants(t *testing.T) { // Its state is made as the grants part makes a declared state, by its stream's name (CreateKeyValue would ask the // account's info, which no module is granted). if _, err := grants.CreateStream(ctx, jetstream.StreamConfig{Name: "KV_" + buscompose.GrantsUserListBucket, - Subjects: []string{"$KV." + buscompose.GrantsUserListBucket + ".>"}, MaxMsgsPerSubject: 1, AllowDirect: true, + Subjects: []string{"$KV." + buscompose.GrantsUserListBucket + ".>"}, MaxMsgsPerSubject: 20, AllowDirect: true, AllowRollup: true, DenyDelete: true, Discard: jetstream.DiscardNew}); err != nil { t.Fatalf("the grants part's state: %v", err) } @@ -140,6 +141,39 @@ func theBusNodeFollowsTheSealedListUnderTheComposedGrants(t *testing.T) { t.Fatal(err) } + // Issue 1966: the grants part purges its sealed lists' keys under its composed grants: listed with their last operations, + // each purged with its value, its state's bytes read before and after. + purging := ownPurging{ctx: ctx, js: grants, module: buscompose.GrantsModule} + live, marked := grantsPart.SealedUserKey("sha256:d9", "a.mod-x"), grantsPart.SealedUserKey("sha256:d8", "a.mod-x") + if _, err := kv.Put(ctx, live, bytes.Repeat([]byte("x"), 4096)); err != nil { + t.Fatal(err) + } + if _, err := kv.Put(ctx, marked, []byte(`{}`)); err != nil { + t.Fatal(err) + } + if err := kv.Delete(ctx, marked); err != nil { + t.Fatal(err) + } + ops, err := purging.LastOps(grantsPart.StateUserList, grantsPart.KeySealedGrants+".>") + if err != nil || len(ops) != 2 || ops[live] || !ops[marked] { + t.Fatalf("the grants part lists its sealed keys with their last operations: %v, %v", ops, err) + } + before, err := purging.Bytes(grantsPart.StateUserList) + if err != nil { + t.Fatalf("the grants part reads its state's bytes: %v", err) + } + for _, k := range []string{live, marked} { + if err := purging.Purge(grantsPart.StateUserList, k); err != nil { + t.Fatalf("the grants part purges its own %s: %v", k, err) + } + } + if after, err := purging.Bytes(grantsPart.StateUserList); err != nil || after+2048 > before { // the value goes; a purge marker of the key stays + t.Errorf("a purge kept the value: %d bytes before, %d after (%v)", before, after, err) + } + if _, err := kv.Get(ctx, live); !errors.Is(err, jetstream.ErrKeyNotFound) { + t.Errorf("the purged %s is still read: %v", live, err) + } + // The bus node's engine: the key watched, the object by name (as mesh-host's link reads a declaration's file). busNode := dial("node.novox") js, err := busNode.JetStream(nats.MaxWait(5 * time.Second)) diff --git a/modules/mesh-grants/cmd/mesh-grants/main.go b/modules/mesh-grants/cmd/mesh-grants/main.go index 2e86973f..f5d37aa6 100644 --- a/modules/mesh-grants/cmd/mesh-grants/main.go +++ b/modules/mesh-grants/cmd/mesh-grants/main.go @@ -128,7 +128,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro hearViews := views.Hear(ctx, byState, logger.Printf) f.Part = &grants.Part{Read: fv, Call: conn, Store: ownStore{conn}, Emit: emitting{conn}, Views: fv, FirstViewContext: ctx, Create: creating{ctx: ctx, js: js}, Consumers: creating{ctx: ctx, js: js}, - Objects: sealedObjects{ctx: ctx, js: js}, Logf: logger.Printf, + Objects: sealedObjects{ctx: ctx, js: js}, Purging: ownPurging{ctx: ctx, js: js, module: account.Module}, Logf: logger.Printf, Bus: grants.BusAddress{URL: account.URL, Fingerprint: account.Fingerprint}} f.Watch = fv // The seat list: read over the connection and the hub, given to catalogue at every change, and a pass composes again then. diff --git a/modules/mesh-grants/cmd/mesh-grants/purge.go b/modules/mesh-grants/cmd/mesh-grants/purge.go new file mode 100644 index 00000000..525f0883 --- /dev/null +++ b/modules/mesh-grants/cmd/mesh-grants/purge.go @@ -0,0 +1,86 @@ +package main + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/nats-io/nats.go/jetstream" +) + +// listWithin bounds one listing of a state's keys with their last operations: a watch of every key of a prefix, which +// for the state user-list's thousands of keys takes longer than one request. +const listWithin = time.Minute + +// ownPurging takes keys of the part's own state away whole (internal/grants purge.go, issue 1966), on the process's +// JetStream connection of its own account. A purge is a publish to the key's own subject, `$KV..`, with the +// purge and roll-up headers: the subject the account already writes and deletes the key on (stateGrants' `$KV..>` +// for a state it keeps). Listing is an ordered consumer filtered on the prefix and the bucket's stream information, both +// in readGrants of its own bucket. +type ownPurging struct { + ctx context.Context + js jetstream.JetStream + module string +} + +func (o ownPurging) kv(ctx context.Context, state string) (jetstream.KeyValue, error) { + kv, err := o.js.KeyValue(ctx, bucketOf(o.module+"."+state)) + if err != nil { + return nil, fmt.Errorf("the bucket %s cannot be opened: %w", bucketOf(o.module+"."+state), err) + } + return kv, nil +} + +func (o ownPurging) Purge(state, key string) error { + ctx, cancel := context.WithTimeout(o.ctx, requestWithin) + defer cancel() + kv, err := o.kv(ctx, state) + if err != nil { + return err + } + return kv.Purge(ctx, key) +} + +func (o ownPurging) LastOps(state, pattern string) (map[string]bool, error) { + ctx, cancel := context.WithTimeout(o.ctx, listWithin) + defer cancel() + kv, err := o.kv(ctx, state) + if err != nil { + return nil, err + } + w, err := kv.Watch(ctx, pattern, jetstream.MetaOnly()) + if err != nil { + return nil, fmt.Errorf("the keys %s cannot be watched: %w", pattern, err) + } + defer func() { _ = w.Stop() }() + out := map[string]bool{} + for { + select { + case <-ctx.Done(): + return nil, fmt.Errorf("the keys %s were not all handed in %s: %w", pattern, listWithin, ctx.Err()) + case e, ok := <-w.Updates(): + if !ok { + return nil, errors.New("the watch of the keys " + pattern + " ended before it handed them all") + } + if e == nil { + return out, nil + } + out[e.Key()] = e.Operation() != jetstream.KeyValuePut + } + } +} + +func (o ownPurging) Bytes(state string) (uint64, error) { + ctx, cancel := context.WithTimeout(o.ctx, requestWithin) + defer cancel() + kv, err := o.kv(ctx, state) + if err != nil { + return 0, err + } + st, err := kv.Status(ctx) + if err != nil { + return 0, err + } + return st.Bytes(), nil +} diff --git a/modules/mesh-grants/grants_test.go b/modules/mesh-grants/grants_test.go index 09d41dd1..0c53614b 100644 --- a/modules/mesh-grants/grants_test.go +++ b/modules/mesh-grants/grants_test.go @@ -3633,6 +3633,8 @@ func TestGrants(t *testing.T) { UserList: &nodeengine.UserListWritten{Source: "a.nats.src2", Revision: 7, Digest: "sha256:aa", Users: 5, Key: busKey.PublicKey}}) p := partOn(b) p.Objects = &fakeObjects{} + purging := &fakePurging{b: b} + p.Purging = purging var digests []string for gen := 1; gen <= 3; gen++ { b.put(grants.ReadAccounts, "a.mod-x", map[string]any{"user": "a.mod-x", "hash": fmt.Sprintf("$2a$10$gen%d", gen), "generation": gen, "fingerprint": fmt.Sprintf("sha256:%d", gen)}) @@ -3656,6 +3658,15 @@ func TestGrants(t *testing.T) { t.Errorf("list %d's a.mod-x key kept = %v", i, kept) } } + // Issue 1966: a delete left every value of the oldest list's keys in the bucket's stream; they are purged, never deleted. + if !slices.Contains(purging.purged, grants.SealedUserKey(digests[0], "a.mod-x")) || slices.Contains(b.writes, grants.StateUserList+" delete") { + t.Errorf("the oldest list's keys were not purged: purged %v, writes %v", purging.purged, b.writes) + } + for _, k := range purging.purged { + if strings.HasPrefix(k, grants.SealedUserKey(digests[1], "")) || strings.HasPrefix(k, grants.SealedUserKey(digests[2], "")) { + t.Errorf("a kept list's key %s was purged", k) + } + } for _, u := range idx.Lists[0].Users { if _, _, ok, _ := b.ownGet(grants.StateUserList, grants.SealedUserKey(digests[2], u)); !ok { t.Errorf("the index names %s, whose key is not there", u) @@ -3929,6 +3940,57 @@ func TestGrants(t *testing.T) { }) } +// fakePurging is the bus's purge of the part's own keys (issue 1966): the key and every value go; marked are the keys whose +// last operation is a delete or purge marker, which a key listing leaves out. +type fakePurging struct { + b *fakeBus + marked map[string]bool + purged []string +} + +func (f *fakePurging) Purge(state, key string) error { + f.b.mu.Lock() + defer f.b.mu.Unlock() + delete(f.b.own[state], key) + delete(f.marked, key) + f.purged = append(f.purged, key) + return nil +} + +func (f *fakePurging) LastOps(state, pattern string) (map[string]bool, error) { + f.b.mu.Lock() + defer f.b.mu.Unlock() + names := func(k string) bool { + if pre, ok := strings.CutSuffix(pattern, ">"); ok { + return strings.HasPrefix(k, pre) + } + return k == pattern + } + out := map[string]bool{} + for k := range f.b.own[state] { + if names(k) { + out[k] = false + } + } + for k := range f.marked { + if names(k) { + out[k] = true + } + } + return out, nil +} + +// Bytes is each value's length and 100 bytes for each marker of the state user-list. +func (f *fakePurging) Bytes(state string) (uint64, error) { + f.b.mu.Lock() + defer f.b.mu.Unlock() + n := uint64(100 * len(f.marked)) + for _, v := range f.b.own[state] { + n += uint64(len(v)) + } + return n, nil +} + // fakeObjects is the part's object bucket of sealed lists. type fakeObjects struct{ objects map[string][]byte } diff --git a/modules/mesh-grants/internal/grants/grants.go b/modules/mesh-grants/internal/grants/grants.go index 25fc80d8..96eee602 100644 --- a/modules/mesh-grants/internal/grants/grants.go +++ b/modules/mesh-grants/internal/grants/grants.go @@ -132,7 +132,7 @@ func SealedUserKey(digest, user string) string { } // keepSealedGrants records the permissions of the list sealed now, one key per user, beside the list before; the lists -// older than those two lose their keys. +// older than those two lose their keys, purged with every value they kept (purge.go: a delete left them in the bucket). func (p *Part) keepSealedGrants(res *Result, digest string, list []byte, composed []User) { in := listUsers(list) var users []User @@ -176,13 +176,13 @@ func (p *Part) keepSealedGrants(res *Result, digest string, list []byte, compose } for _, l := range gone { for _, user := range l.Users { - if err := p.Store.Delete(StateUserList, SealedUserKey(l.Digest, user)); err != nil { + if err := p.takeAway(StateUserList, SealedUserKey(l.Digest, user)); err != nil { res.Errors = append(res.Errors, fmt.Sprintf("its own %s %s cannot be taken away: %v", StateUserList, SealedUserKey(l.Digest, user), err)) } } } if _, _, found, err := p.Store.Get(StateUserList, KeySealedGrants); err == nil && found { - if err := p.Store.Delete(StateUserList, KeySealedGrants); err != nil { + if err := p.takeAway(StateUserList, KeySealedGrants); err != nil { res.Errors = append(res.Errors, fmt.Sprintf("its own %s %s cannot be taken away: %v", StateUserList, KeySealedGrants, err)) } } diff --git a/modules/mesh-grants/internal/grants/part.go b/modules/mesh-grants/internal/grants/part.go index 6590b9ec..5e4ea403 100644 --- a/modules/mesh-grants/internal/grants/part.go +++ b/modules/mesh-grants/internal/grants/part.go @@ -158,6 +158,9 @@ type Part struct { Logf func(string, ...any) // Objects is the part's object bucket of sealed lists: the process's connection; nil in the tools. Objects Objects + // Purging takes keys of its own state away whole, values and all (purge.go): the process's connection; nil in the + // tools, where a key taken away is deleted. + Purging Purger // Consumers creates an account's durable consumer (consumers.go): the process's connection; nil in the tools, which // compare only. Consumers ConsumerCreator diff --git a/modules/mesh-grants/internal/grants/purge.go b/modules/mesh-grants/internal/grants/purge.go new file mode 100644 index 00000000..12f09d94 --- /dev/null +++ b/modules/mesh-grants/internal/grants/purge.go @@ -0,0 +1,23 @@ +package grants + +// Purger takes keys of the part's own state away whole (issue 1966, 2026-11-08): a key-value delete only adds a marker +// and leaves every value the key kept in the bucket's stream, so the per-user keys of the sealed lists older than the +// two kept filled the state user-list to 94% of its cap (5,968 keys, 13,228 values kept, 23:20 UTC). A purge takes the +// key's values with it, leaving one marker. The process's own connection; nil in the tools, where a key taken away is deleted. +type Purger interface { + // Purge takes key of state away with every value it kept. + Purge(state, key string) error + // LastOps is every key of state that pattern names (one key, or a prefix ending `.>`), each with whether its last + // operation is a delete or purge marker: the keys a key listing leaves out. + LastOps(state, pattern string) (map[string]bool, error) + // Bytes is how many bytes the state holds on the bus, values and markers. + Bytes(state string) (uint64, error) +} + +// takeAway takes a key of the part's own state away: purged where the part holds a purger, else deleted. +func (p *Part) takeAway(state, key string) error { + if p.Purging != nil { + return p.Purging.Purge(state, key) + } + return p.Store.Delete(state, key) +}