Merge pull request 'Grants purges old sealed lists' keys, and once at its start the ones a delete left behind (issue 1966, recovery)' (#182) from fix/1966-user-list-purge into main
This commit was merged in pull request #182.
This commit is contained in:
@@ -7,6 +7,7 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"encoding/base64"
|
"encoding/base64"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"slices"
|
"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
|
// 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).
|
// account's info, which no module is granted).
|
||||||
if _, err := grants.CreateStream(ctx, jetstream.StreamConfig{Name: "KV_" + buscompose.GrantsUserListBucket,
|
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 {
|
AllowRollup: true, DenyDelete: true, Discard: jetstream.DiscardNew}); err != nil {
|
||||||
t.Fatalf("the grants part's state: %v", err)
|
t.Fatalf("the grants part's state: %v", err)
|
||||||
}
|
}
|
||||||
@@ -140,6 +141,39 @@ func theBusNodeFollowsTheSealedListUnderTheComposedGrants(t *testing.T) {
|
|||||||
t.Fatal(err)
|
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).
|
// 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")
|
busNode := dial("node.novox")
|
||||||
js, err := busNode.JetStream(nats.MaxWait(5 * time.Second))
|
js, err := busNode.JetStream(nats.MaxWait(5 * time.Second))
|
||||||
|
|||||||
@@ -128,7 +128,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
|||||||
hearViews := views.Hear(ctx, byState, logger.Printf)
|
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,
|
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},
|
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}}
|
Bus: grants.BusAddress{URL: account.URL, Fingerprint: account.Fingerprint}}
|
||||||
f.Watch = fv
|
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.
|
// The seat list: read over the connection and the hub, given to catalogue at every change, and a pass composes again then.
|
||||||
@@ -207,6 +207,17 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
|||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
// Once, before the first pass writes a list's keys: the sealed lists' keys a delete left in the state user-list are
|
||||||
|
// purged (issue 1966), said with the count and the bytes before and after.
|
||||||
|
if sw, err := f.Part.SweepSealed(); err != nil {
|
||||||
|
logger.Printf("the sealed lists' keys of %s were not swept: %v", grants.StateUserList, err)
|
||||||
|
} else {
|
||||||
|
for _, e := range sw.Errors {
|
||||||
|
logger.Printf("%s", e)
|
||||||
|
}
|
||||||
|
logger.Printf("swept the sealed lists' keys of %s: %d marked by a delete and %d of lists no longer kept purged, the kept lists %v untouched (no index: %v); %d bytes before, %d after",
|
||||||
|
grants.StateUserList, sw.Marked, sw.Orphaned, sw.Kept, sw.NoIndex, sw.BytesBefore, sw.BytesAfter)
|
||||||
|
}
|
||||||
logger.Printf("started on the account %s: publishing memberships, writing grants and the user list sealed to the bus node, which it follows, and creating the account consumers (the heard-stream ones too) and the declared buckets, logs and streams the bus lacks; a held stream and a held consumer only ever gain the subjects and filters a manifest adds", account.User)
|
logger.Printf("started on the account %s: publishing memberships, writing grants and the user list sealed to the bus node, which it follows, and creating the account consumers (the heard-stream ones too) and the declared buckets, logs and streams the bus lacks; a held stream and a held consumer only ever gain the subjects and filters a manifest adds", account.User)
|
||||||
f.Run(ctx)
|
f.Run(ctx)
|
||||||
return nil
|
return nil
|
||||||
|
|||||||
@@ -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.<bucket>.<key>`, with the
|
||||||
|
// purge and roll-up headers: the subject the account already writes and deletes the key on (stateGrants' `$KV.<bucket>.>`
|
||||||
|
// 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
|
||||||
|
}
|
||||||
@@ -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}})
|
UserList: &nodeengine.UserListWritten{Source: "a.nats.src2", Revision: 7, Digest: "sha256:aa", Users: 5, Key: busKey.PublicKey}})
|
||||||
p := partOn(b)
|
p := partOn(b)
|
||||||
p.Objects = &fakeObjects{}
|
p.Objects = &fakeObjects{}
|
||||||
|
purging := &fakePurging{b: b}
|
||||||
|
p.Purging = purging
|
||||||
var digests []string
|
var digests []string
|
||||||
for gen := 1; gen <= 3; gen++ {
|
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)})
|
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)
|
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 {
|
for _, u := range idx.Lists[0].Users {
|
||||||
if _, _, ok, _ := b.ownGet(grants.StateUserList, grants.SealedUserKey(digests[2], u)); !ok {
|
if _, _, ok, _ := b.ownGet(grants.StateUserList, grants.SealedUserKey(digests[2], u)); !ok {
|
||||||
t.Errorf("the index names %s, whose key is not there", u)
|
t.Errorf("the index names %s, whose key is not there", u)
|
||||||
@@ -3671,6 +3682,79 @@ func TestGrants(t *testing.T) {
|
|||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
|
||||||
|
// Issue 1966, 2026-11-08 23:20: the state user-list held 5,968 keys, most of them markers of deletes whose values the
|
||||||
|
// bucket kept (94% of its cap). Once at the process's start, the marked sealed-grants keys and the keys of lists the
|
||||||
|
// index no longer names are purged; a kept list's keys and every other key of the state stay.
|
||||||
|
t.Run("sealed sweep: purges marked sealed-grants keys and lists no longer kept, never a kept list's or another key", func(t *testing.T) {
|
||||||
|
const d0, d1, d2, d9 = "sha256:d0", "sha256:d1", "sha256:d2", "sha256:d9"
|
||||||
|
seed := func(index bool) (*fakeBus, *fakePurging) {
|
||||||
|
b := fixture()
|
||||||
|
pg := &fakePurging{b: b, marked: map[string]bool{}}
|
||||||
|
if index {
|
||||||
|
raw, _ := json.Marshal(grants.SealedLists{Lists: []grants.SealedNames{{Digest: d2, Users: []string{"a.mod-x"}}, {Digest: d1, Users: []string{"a.mod-x"}}}})
|
||||||
|
_, _ = ownStore{b}.Put(grants.StateUserList, grants.KeySealedLists, raw)
|
||||||
|
}
|
||||||
|
for _, k := range []string{grants.SealedUserKey(d2, "a.mod-x"), grants.SealedUserKey(d1, "a.mod-x"), grants.SealedUserKey(d0, "a.mod-x"),
|
||||||
|
grants.SealedUserKey(d0, "b.mod-y"), "pass", grants.KeyWrittenCredentials} {
|
||||||
|
_, _ = ownStore{b}.Put(grants.StateUserList, k, []byte(`{"user":"a.mod-x","pad":"`+strings.Repeat("x", 4800)+`"}`))
|
||||||
|
}
|
||||||
|
// Markers: a deleted list's keys, the one value written before the index, a kept list's key and another key.
|
||||||
|
for _, k := range []string{grants.SealedUserKey(d9, "a.mod-x"), grants.SealedUserKey(d9, "b.mod-y"), grants.KeySealedGrants,
|
||||||
|
grants.SealedUserKey(d1, "b.mod-y"), grants.KeyHashHold} {
|
||||||
|
pg.marked[k] = true
|
||||||
|
}
|
||||||
|
return b, pg
|
||||||
|
}
|
||||||
|
rows := []struct {
|
||||||
|
name string
|
||||||
|
index, refuse bool
|
||||||
|
marked, orphaned int
|
||||||
|
purged, untouched []string
|
||||||
|
err string
|
||||||
|
}{
|
||||||
|
{name: "with the index", index: true, marked: 3, orphaned: 2,
|
||||||
|
purged: []string{grants.SealedUserKey(d9, "a.mod-x"), grants.SealedUserKey(d9, "b.mod-y"), grants.KeySealedGrants,
|
||||||
|
grants.SealedUserKey(d0, "a.mod-x"), grants.SealedUserKey(d0, "b.mod-y")}},
|
||||||
|
{name: "no index: the marked keys alone, no list kept", marked: 4,
|
||||||
|
purged: []string{grants.SealedUserKey(d9, "a.mod-x"), grants.SealedUserKey(d9, "b.mod-y"), grants.KeySealedGrants,
|
||||||
|
grants.SealedUserKey(d1, "b.mod-y")},
|
||||||
|
untouched: []string{grants.SealedUserKey(d0, "a.mod-x"), grants.SealedUserKey(d0, "b.mod-y")}},
|
||||||
|
{name: "an unreadable index: nothing", index: true, refuse: true, err: "index of kept lists cannot be read"},
|
||||||
|
}
|
||||||
|
for _, r := range rows {
|
||||||
|
b, pg := seed(r.index)
|
||||||
|
p := partOn(b)
|
||||||
|
p.Purging = pg
|
||||||
|
if r.refuse {
|
||||||
|
p.Store = refusingIndex{ownStore{b}}
|
||||||
|
}
|
||||||
|
sw, err := p.SweepSealed()
|
||||||
|
if r.err != "" {
|
||||||
|
if err == nil || !strings.Contains(err.Error(), r.err) || len(pg.purged) > 0 {
|
||||||
|
t.Errorf("%s: err %v, purged %v", r.name, err, pg.purged)
|
||||||
|
}
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if err != nil || sw.Marked != r.marked || sw.Orphaned != r.orphaned || sw.NoIndex == r.index || sw.BytesAfter >= sw.BytesBefore {
|
||||||
|
t.Errorf("%s: %+v, %v", r.name, sw, err)
|
||||||
|
}
|
||||||
|
if !slices.Equal(slices.Sorted(slices.Values(pg.purged)), slices.Sorted(slices.Values(r.purged))) {
|
||||||
|
t.Errorf("%s: purged %v, want %v", r.name, pg.purged, r.purged)
|
||||||
|
}
|
||||||
|
for _, k := range append([]string{grants.SealedUserKey(d2, "a.mod-x"), grants.SealedUserKey(d1, "a.mod-x"), "pass", grants.KeyWrittenCredentials}, r.untouched...) {
|
||||||
|
if _, _, ok, _ := b.ownGet(grants.StateUserList, k); !ok {
|
||||||
|
t.Errorf("%s: %s was taken away", r.name, k)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (r.index && !pg.marked[grants.SealedUserKey(d1, "b.mod-y")]) || !pg.marked[grants.KeyHashHold] {
|
||||||
|
t.Errorf("%s: a kept list's or another key's marker was purged: %v", r.name, pg.purged)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if _, err := partOn(fixture()).SweepSealed(); err == nil {
|
||||||
|
t.Error("a part with no purger swept")
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
// Sweep 2026-10-30 item 5, a kind met twice that day (the engine's acts, then mesh-host_health): one principal the
|
// Sweep 2026-10-30 item 5, a kind met twice that day (the engine's acts, then mesh-host_health): one principal the
|
||||||
// composer refuses as a second writer froze the whole list. Now it keeps its grants of the list in force, the rest is
|
// composer refuses as a second writer froze the whole list. Now it keeps its grants of the list in force, the rest is
|
||||||
// written, and the refusal is raised by name; one the list in force does not carry stays out, said too.
|
// written, and the refusal is raised by name; one the list in force does not carry stays out, said too.
|
||||||
@@ -3929,6 +4013,67 @@ 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
|
||||||
|
}
|
||||||
|
|
||||||
|
// refusingIndex is the part's own state with the index of kept lists unreadable.
|
||||||
|
type refusingIndex struct{ ownStore }
|
||||||
|
|
||||||
|
func (r refusingIndex) Get(state, key string) ([]byte, uint64, bool, error) {
|
||||||
|
if key == grants.KeySealedLists {
|
||||||
|
return nil, 0, false, errors.New("nats: timeout")
|
||||||
|
}
|
||||||
|
return r.ownStore.Get(state, key)
|
||||||
|
}
|
||||||
|
|
||||||
// fakeObjects is the part's object bucket of sealed lists.
|
// fakeObjects is the part's object bucket of sealed lists.
|
||||||
type fakeObjects struct{ objects map[string][]byte }
|
type fakeObjects struct{ objects map[string][]byte }
|
||||||
|
|
||||||
|
|||||||
@@ -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
|
// 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) {
|
func (p *Part) keepSealedGrants(res *Result, digest string, list []byte, composed []User) {
|
||||||
in := listUsers(list)
|
in := listUsers(list)
|
||||||
var users []User
|
var users []User
|
||||||
@@ -176,13 +176,13 @@ func (p *Part) keepSealedGrants(res *Result, digest string, list []byte, compose
|
|||||||
}
|
}
|
||||||
for _, l := range gone {
|
for _, l := range gone {
|
||||||
for _, user := range l.Users {
|
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))
|
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 _, _, 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))
|
res.Errors = append(res.Errors, fmt.Sprintf("its own %s %s cannot be taken away: %v", StateUserList, KeySealedGrants, err))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -158,6 +158,9 @@ type Part struct {
|
|||||||
Logf func(string, ...any)
|
Logf func(string, ...any)
|
||||||
// Objects is the part's object bucket of sealed lists: the process's connection; nil in the tools.
|
// Objects is the part's object bucket of sealed lists: the process's connection; nil in the tools.
|
||||||
Objects Objects
|
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
|
// Consumers creates an account's durable consumer (consumers.go): the process's connection; nil in the tools, which
|
||||||
// compare only.
|
// compare only.
|
||||||
Consumers ConsumerCreator
|
Consumers ConsumerCreator
|
||||||
|
|||||||
@@ -0,0 +1,114 @@
|
|||||||
|
package grants
|
||||||
|
|
||||||
|
import (
|
||||||
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"strings"
|
||||||
|
)
|
||||||
|
|
||||||
|
// 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)
|
||||||
|
}
|
||||||
|
|
||||||
|
// sealedPrefix is the prefix of every user's key of the sealed list of digest.
|
||||||
|
func sealedPrefix(digest string) string { return SealedUserKey(digest, "") }
|
||||||
|
|
||||||
|
// SealedSweep is what SweepSealed took away of the state user-list.
|
||||||
|
type SealedSweep struct {
|
||||||
|
// Marked are the sealed-grants keys whose last operation was a delete or purge marker, purged; Orphaned the live keys
|
||||||
|
// of a list the index of kept lists does not name, purged.
|
||||||
|
Marked int `json:"marked"`
|
||||||
|
Orphaned int `json:"orphaned"`
|
||||||
|
// Kept is the digests of the kept lists, whose keys the sweep never touches.
|
||||||
|
Kept []string `json:"kept"`
|
||||||
|
// NoIndex says the state holds no index of kept lists: then only the marked keys are purged.
|
||||||
|
NoIndex bool `json:"no_index,omitempty"`
|
||||||
|
BytesBefore uint64 `json:"bytes_before"`
|
||||||
|
BytesAfter uint64 `json:"bytes_after"`
|
||||||
|
Errors []string `json:"errors,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// SweepSealed purges, once, the keys of the sealed lists that a delete left in the state user-list (issue 1966): every
|
||||||
|
// sealed-grants key whose last operation is a delete or purge marker, and every user's key of a list the index of kept
|
||||||
|
// lists does not name. Never a key of a kept list, and never another key of the state. Run at the process's start,
|
||||||
|
// before its first pass, so no pass writes a list's keys beside it. Refused whole when the index cannot be read.
|
||||||
|
func (p *Part) SweepSealed() (SealedSweep, error) {
|
||||||
|
var out SealedSweep
|
||||||
|
if p.Purging == nil {
|
||||||
|
return out, errors.New("the part holds no purger: nothing is taken away")
|
||||||
|
}
|
||||||
|
raw, _, found, err := p.Store.Get(StateUserList, KeySealedLists)
|
||||||
|
if err != nil {
|
||||||
|
return out, fmt.Errorf("the index of kept lists cannot be read, so which keys are kept is not known: %w", err)
|
||||||
|
}
|
||||||
|
var idx SealedLists
|
||||||
|
if found {
|
||||||
|
if err := json.Unmarshal(raw, &idx); err != nil {
|
||||||
|
return out, fmt.Errorf("the index of kept lists cannot be read, so which keys are kept is not known: %w", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
out.NoIndex = !found
|
||||||
|
var kept []string
|
||||||
|
for _, l := range idx.Lists {
|
||||||
|
out.Kept = append(out.Kept, l.Digest)
|
||||||
|
kept = append(kept, sealedPrefix(l.Digest))
|
||||||
|
}
|
||||||
|
if out.BytesBefore, err = p.Purging.Bytes(StateUserList); err != nil {
|
||||||
|
return out, fmt.Errorf("the state %s cannot be read: %w", StateUserList, err)
|
||||||
|
}
|
||||||
|
isKept := func(key string) bool {
|
||||||
|
for _, pre := range kept {
|
||||||
|
if strings.HasPrefix(key, pre) {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
purge := func(key string, n *int) {
|
||||||
|
if err := p.Purging.Purge(StateUserList, key); err != nil {
|
||||||
|
out.Errors = append(out.Errors, fmt.Sprintf("its own %s %s cannot be purged: %v", StateUserList, key, err))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
*n++
|
||||||
|
}
|
||||||
|
// The one value written before the index (KeySealedGrants), once deleted, and every user's key.
|
||||||
|
for _, pattern := range []string{KeySealedGrants, KeySealedGrants + ".>"} {
|
||||||
|
ops, err := p.Purging.LastOps(StateUserList, pattern)
|
||||||
|
if err != nil {
|
||||||
|
return out, fmt.Errorf("the keys %s of %s cannot be listed: %w", pattern, StateUserList, err)
|
||||||
|
}
|
||||||
|
for _, key := range sortedKeys(ops) {
|
||||||
|
switch {
|
||||||
|
case key != KeySealedGrants && !strings.HasPrefix(key, KeySealedGrants+"."):
|
||||||
|
case key != KeySealedGrants && isKept(key):
|
||||||
|
case ops[key]:
|
||||||
|
purge(key, &out.Marked)
|
||||||
|
case key != KeySealedGrants && found:
|
||||||
|
purge(key, &out.Orphaned)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if out.BytesAfter, err = p.Purging.Bytes(StateUserList); err != nil {
|
||||||
|
out.Errors = append(out.Errors, fmt.Sprintf("the state %s cannot be read after the sweep: %v", StateUserList, err))
|
||||||
|
}
|
||||||
|
return out, nil
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user