novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request 'Registry recovery: break the circle, and let a cut part be sent again (issue 1966)' (#204) from fix/1966-registry-circle into main

Browse Source
This commit was merged in pull request #204.
This commit is contained in:
mesh-tools
2026-11-11 13:18:27 +00:00
parent 8f5cbce468 7db5b0dfe4
commit 2f1b60f2a2
5 changed files with 173 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-grants/cmd/mesh-grants/integration_test.go
+28
View File
@@ -224,6 +224,8 @@ func TestTheGrantsPartWhole(t *testing.T) {
type listGate struct{ asked, release chan struct{} }
var gate atomic.Pointer[listGate]
var answeredGated, listAsked, listAnswered atomic.Int32
// While set, the bus module answers as a build from before the field `binding` (design 25 bindings-first-view §9).
var oldBuild atomic.Bool
for _, subject := range []string{"mesh.seat.mesh-broker.call.list.>", "mesh.seat.mesh-broker.tool.list"} {
if _, err := nc.Subscribe(subject, func(m *nats.Msg) {
if !listing.Load() {
@@ -233,6 +235,15 @@ func TestTheGrantsPartWhole(t *testing.T) {
listAsked.Add(1)
defer listAnswered.Add(1)
var more []any
if oldBuild.Load() {
answer(m, map[string]any{"entries": []any{
map[string]any{"module": "mod-c", "local-name": "late", "storage-type": "key-value", "template": "state", "bus-name": "mod-c_late",
"settings": map[string]any{"KV_mod-c_late": map[string]any{}}},
map[string]any{"module": "mod-c", "local-name": "feed", "storage-type": "stream", "bus-name": "mod-c_feed",
"settings": map[string]any{"mod-c_feed": map[string]any{}}},
}})
return
}
if g := gate.Load(); g != nil {
g.asked <- struct{}{}
<-g.release
@@ -440,6 +451,23 @@ func TestTheGrantsPartWhole(t *testing.T) {
if b := bindingOf(); len(b) != 0 {
t.Fatalf("after an answer that lists a binding dropped since its ask: %+v", b)
}
// 1966/AC203-3..: a bus module's build from before the field answers no binding on any entry (design 25 bindings-first-view
// §9): the key-value entry the bus holds gets its binding composed from the answer's fields, the stream is left unbound.
oldBuild.Store(true)
askedAgain, _ = json.Marshal(grants.Ask{By: "someone@desk", At: time.Now().Add(2 * time.Second), FirstView: grants.FirstViewBindings})
if _, err := kvs[grants.Module+"."+grants.StateUserList].Put(ctx, grants.KeyAsked, askedAgain); err != nil {
t.Fatal(err)
}
until("the binding of the entry composed from an answer with no binding, the stream left unbound", func() bool {
e, err := kvs[grants.Module+"."+grants.StateUserList].Get(ctx, grants.KeyFirstViews)
if err != nil {
return false
}
var st grants.FirstViewsState
b := bindingOf()
return json.Unmarshal(e.Value(), &st) == nil && st.Bindings != nil && st.Bindings.Composed == 1 &&
slices.Equal(st.Bindings.Unbound, []string{"mod-c/feed (stream)"}) && len(b) == 1 && b[0].Bucket == "mod-c_late"
})
for _, bucket := range []string{"mesh-controller_lease", grants.Module + "_lease", grants.Module + "_takeover"} {
if _, err := js.KeyValue(ctx, bucket); !errors.Is(err, jetstream.ErrBucketNotFound) {
t.Errorf("the bucket %s: %v, want none", bucket, err)
modules/mesh-grants/grants_test.go
+61 -20
View File
@@ -3064,8 +3064,16 @@ func TestGrants(t *testing.T) {
"binding": map[string]any{"local": local, "storage": "key-value", "template": "state", "bucket": bucket}}
}
withoutBinding := map[string]any{"module": "mod-x", "local-name": "waiting", "storage-type": "key-value", "bus-name": "mod-x_waiting", "state": "waiting"}
oldBusModule := map[string]any{"module": "mod-x", "local-name": "late", "storage-type": "key-value", "bus-name": "mod-x_late",
"settings": map[string]any{"mod-x_late": map[string]any{}}}
// Design 25 bindings-first-view §9: the entries of a bus module's answer from before the field, with no binding. An
// entry carries what the bus holds in settings, keyed by the bus object's name (KV_ and OBJ_ before the bus name).
noBinding := func(local, storage, object string, held bool) map[string]any {
e := map[string]any{"module": "mod-x", "local-name": local, "storage-type": storage, "template": "state", "bus-name": "mod-x_" + local}
if held {
e["settings"] = map[string]any{object + "mod-x_" + local: map[string]any{}}
}
return e
}
oldBusModule := noBinding("late", "key-value", "KV_", true)
// A bus module's setup: the part with the first view, the conditions it said, the bindings it keeps for mod-x.
type setup struct {
b *fakeBus
@@ -3109,7 +3117,7 @@ func TestGrants(t *testing.T) {
var bound grants.Bound
_ = json.Unmarshal(raw, &bound)
for local, b := range bound.Bindings {
out[local] = b.Bucket
out[local] = b.Bucket + b.ObjectStore
}
}
return out
@@ -3122,17 +3130,21 @@ func TestGrants(t *testing.T) {
}
}
for _, row := range []struct {
name string
held map[string]string // kept before, by local name
answer []any
during func(*setup) // heard while the ask runs
want map[string]string
status [3]int // written, rewritten, listed
said map[string]string
legacy bool
again []any // a second answer
wantAgn map[string]string
saidAgn map[string]string
name string
held map[string]string // kept before, by local name
answer []any
during func(*setup) // heard while the ask runs
want map[string]string
status [3]int // written, rewritten, listed
said map[string]string
// 1966/AC203-3..: §9's composed count and unbound entries after the last answer, and how many answers said
// that they carry no binding on any entry.
composed int
unbound []string
noted int
again []any // a second answer
wantAgn map[string]string
saidAgn map[string]string
}{
{name: "a binding the events never brought is written", answer: []any{entry("late", "mod-x_late")}, want: map[string]string{"late": "mod-x_late"}, status: [3]int{1, 0, 1}},
{name: "the same binding writes nothing", held: map[string]string{"late": "mod-x_late"}, answer: []any{entry("late", "mod-x_late")},
@@ -3140,7 +3152,7 @@ func TestGrants(t *testing.T) {
{name: "a different binding is rewritten", held: map[string]string{"late": "mod-x_old"}, answer: []any{entry("late", "mod-x_late")},
want: map[string]string{"late": "mod-x_late"}, status: [3]int{0, 1, 1}},
{name: "an entry listed without a binding changes nothing", held: map[string]string{"waiting": "mod-x_waiting"}, answer: []any{withoutBinding},
want: map[string]string{"waiting": "mod-x_waiting"}, status: [3]int{0, 0, 1}},
want: map[string]string{"waiting": "mod-x_waiting"}, status: [3]int{0, 0, 1}, unbound: []string{"mod-x/waiting (key-value)"}, noted: 1},
{name: "a binding the answer omits is kept, raises its condition, and a later answer that lists it clears it",
held: map[string]string{"gone": "mod-x_gone"}, answer: []any{}, want: map[string]string{"gone": "mod-x_gone"},
said: map[string]string{grants.BindingUnlisted + ".mod-x.gone": "raised"},
@@ -3151,8 +3163,29 @@ func TestGrants(t *testing.T) {
{name: "a dropped heard since the ask is not written back, and raises nothing", held: map[string]string{"late": "mod-x_late"},
answer: []any{entry("late", "mod-x_late")}, during: func(st *setup) { hear(st, "dropped", "late", "") },
want: map[string]string{}, status: [3]int{0, 0, 1}, said: map[string]string{}},
{name: "an answer of a bus module from before the field writes nothing and says so once", answer: []any{oldBusModule},
want: map[string]string{}, status: [3]int{0, 0, 1}, legacy: true, again: []any{oldBusModule}, wantAgn: map[string]string{}},
// §9 (1966/AC203-3..): an answer no entry of which carries a binding.
{name: "§9 a key-value entry the bus holds is composed from the answer's fields, and said once per answer", answer: []any{oldBusModule},
want: map[string]string{"late": "mod-x_late"}, status: [3]int{1, 0, 1}, composed: 0, noted: 2, again: []any{oldBusModule},
wantAgn: map[string]string{"late": "mod-x_late"}},
{name: "§9 an entry an event named since the ask is not written, so it is not counted composed", answer: []any{oldBusModule},
during: func(st *setup) { hear(st, "created", "late", "mod-x_event") }, want: map[string]string{"late": "mod-x_event"},
status: [3]int{0, 0, 1}, noted: 1},
{name: "§9 an object-store entry is composed with its object store", answer: []any{noBinding("media", "object-store", "OBJ_", true)},
want: map[string]string{"media": "mod-x_media"}, status: [3]int{1, 0, 1}, composed: 1, noted: 1},
{name: "§9 a stream and a ledger are left unbound and logged", answer: []any{noBinding("s", "stream", "", true), noBinding("l", "ledger", "", true)},
want: map[string]string{}, status: [3]int{0, 0, 2}, unbound: []string{"mod-x/l (ledger)", "mod-x/s (stream)"}, noted: 1},
{name: "§9 an entry without settings is left unbound and logged", answer: []any{noBinding("waiting", "key-value", "KV_", false)},
want: map[string]string{}, status: [3]int{0, 0, 1}, unbound: []string{"mod-x/waiting (key-value)"}, noted: 1},
{name: "§9 an entry whose settings lack its bus object is left unbound and logged",
answer: []any{func() map[string]any {
e := noBinding("late", "key-value", "KV_", true)
e["settings"] = map[string]any{"mod-x_late": map[string]any{}}
return e
}()},
want: map[string]string{}, status: [3]int{0, 0, 1}, unbound: []string{"mod-x/late (key-value)"}, noted: 1},
{name: "§9 an answer where any entry carries a binding is applied by §4 alone, nothing composed",
answer: []any{entry("other", "mod-x_other"), oldBusModule},
want: map[string]string{"other": "mod-x_other"}, status: [3]int{1, 0, 2}},
} {
t.Run(row.name, func(t *testing.T) {
st := newSetup()
@@ -3181,7 +3214,7 @@ func TestGrants(t *testing.T) {
t.Fatal(err)
}
check(row.want, row.said)
if s := st.v.Status(); [3]int{s.Written, s.Rewritten, s.Listed} != row.status && !row.legacy && row.again == nil {
if s := st.v.Status(); [3]int{s.Written, s.Rewritten, s.Listed} != row.status && row.again == nil {
t.Errorf("status wrote %d, rewrote %d, listed %d; want %v", s.Written, s.Rewritten, s.Listed, row.status)
}
if row.again != nil {
@@ -3194,8 +3227,16 @@ func TestGrants(t *testing.T) {
t.Errorf("by hand %d, want 1", st.v.Status().ByHand)
}
}
if told := strings.Count(strings.Join(st.logs, "\n"), "no binding (a bus module from before the field)"); row.legacy && told != 1 || !row.legacy && told != 0 {
t.Errorf("a bus module from before the field was told %d times", told)
if told := strings.Count(strings.Join(st.logs, "\n"), "answered with no binding on any entry"); told != row.noted {
t.Errorf("an answer with no binding was told %d times, want %d", told, row.noted)
}
if s := st.v.Status(); s.Composed != row.composed || !slices.Equal(s.Unbound, row.unbound) {
t.Errorf("status composed %d, unbound %v; want %d, %v", s.Composed, s.Unbound, row.composed, row.unbound)
}
for _, u := range row.unbound {
if !strings.Contains(strings.Join(st.logs, "\n"), u) {
t.Errorf("the log does not name the unbound entry %s: %v", u, st.logs)
}
}
})
}
modules/mesh-grants/internal/grants/bindingsview.go
+70 -22
View File
@@ -43,6 +43,11 @@ type BindingsStatus struct {
Listed int `json:"listed"`
Written int `json:"written"`
Rewritten int `json:"rewritten"`
// Composed is how many of the bindings the last answer wrote or rewrote were composed from its own fields, and Unbound
// the entries it left unbound, `<module>/<local name> (<storage type>)`: both only for an answer no entry of which
// carries `binding` (§9).
Composed int `json:"composed"`
Unbound []string `json:"unbound,omitempty"`
// Unlisted are the bindings the part holds that the last answer did not list, `<module>/<local name>`.
Unlisted []string `json:"unlisted,omitempty"`
Asks int `json:"asks"`
@@ -77,7 +82,6 @@ type BindingsView struct {
held []heldEvent
touched map[string]bool
raised map[string]bool // the unlisted conditions raised, by `<module>/<local name>`
legacy bool
}
type heldEvent struct {
@@ -323,27 +327,20 @@ func (v *BindingsView) apply(p *Part, raw json.RawMessage) error {
if err := json.Unmarshal(raw, &answer); err != nil {
return fmt.Errorf("%s answered what is not a list: %w", CallBrokerList, err)
}
entries, composed, unbound := v.compose(answer.Entries)
p.bindMu.Lock()
written, rewritten, holdsNoBinding, err := v.take(p, answer.Entries)
written, rewritten, fromFields, err := v.take(p, entries, composed)
p.bindMu.Unlock()
if err != nil {
return err
}
if holdsNoBinding > 0 {
v.mu.Lock()
said := v.legacy
v.legacy = true
v.mu.Unlock()
if !said {
v.logf("the seat %s's verb %s lists %d entries whose objects the bus holds with no binding (a bus module from before the field): nothing is written from the answer, and the events are heard as before", BrokerSeat, BrokerListVerb, holdsNoBinding)
}
}
v.mu.Lock()
wasRaised := v.st.Raised
v.st.Raised, v.st.Error = false, ""
v.st.Answered = v.now()
v.st.Listed = len(answer.Entries)
v.st.Written, v.st.Rewritten = written, rewritten
v.st.Composed, v.st.Unbound = fromFields, unbound
status := v.statusLocked()
v.mu.Unlock()
if wasRaised {
@@ -354,10 +351,57 @@ func (v *BindingsView) apply(p *Part, raw json.RawMessage) error {
return nil
}
// compose gives the entries of an answer that carries no `binding` on any entry (a bus module's build from before the
// field) the binding its own fields name whole (design 25 bindings-first-view §9): for a key-value entry or an object-store
// entry whose settings hold the bus object its bus name names. It answers the entries with those bindings, the entries
// composed (`<module>/<local name>`) and the ones left unbound, and says once what it did. An answer where any entry carries
// `binding` is applied by §4 alone: it answers the entries as they are.
func (v *BindingsView) compose(entries []meshbroker.Held) ([]meshbroker.Held, map[string]bool, []string) {
if len(entries) == 0 || slices.ContainsFunc(entries, func(e meshbroker.Held) bool { return e.Binding != nil }) {
return entries, nil, nil
}
out := slices.Clone(entries)
composed := map[string]bool{}
var unbound []string
for i, e := range out {
if b, ok := composeBinding(e); ok {
out[i].Binding = b
composed[localKey(e.Module, e.LocalName)] = true
continue
}
unbound = append(unbound, fmt.Sprintf("%s (%s)", localKey(e.Module, e.LocalName), e.StorageType))
}
sort.Strings(unbound)
v.logf("the seat %s's verb %s answered with no binding on any entry (a bus module from before the field): %d bindings composed from its fields, left unbound: %s",
BrokerSeat, BrokerListVerb, len(composed), strings.Join(unbound, ", "))
return out, composed, unbound
}
// composeBinding is §9's binding of one entry: only for the storage types whose binding the answer names whole, and only
// when the entry's settings hold the bus object its bus name names (KV_ and the bus name, OBJ_ and the bus name).
func composeBinding(e meshbroker.Held) (*meshbroker.Binding, bool) {
b := &meshbroker.Binding{Local: e.LocalName, Storage: e.StorageType, Template: e.Template}
switch e.StorageType {
case "key-value":
if _, held := e.Settings["KV_"+e.BusName]; !held {
return nil, false
}
b.Bucket = e.BusName
case "object-store":
if _, held := e.Settings["OBJ_"+e.BusName]; !held {
return nil, false
}
b.ObjectStore = e.BusName
default:
return nil, false
}
return b, e.LocalName != "" && e.BusName != ""
}
// take writes the bindings of the entries an event did not name since the ask was sent, keeps every binding the answer does
// not list and raises its unlisted condition, and answers how many bindings it wrote and rewrote, and how many entries it
// lists with their objects held and no binding. The caller holds the part's bindMu.
func (v *BindingsView) take(p *Part, entries []meshbroker.Held) (written, rewritten, noBinding int, err error) {
// not list and raises its unlisted condition, and answers how many bindings it wrote and rewrote; composed names the
// entries whose binding compose gave them. The caller holds the part's bindMu.
func (v *BindingsView) take(p *Part, entries []meshbroker.Held, composed map[string]bool) (written, rewritten, fromFields int, err error) {
v.mu.Lock()
touched := copyOf(v.touched)
v.mu.Unlock()
@@ -366,9 +410,6 @@ func (v *BindingsView) take(p *Part, entries []meshbroker.Held) (written, rewrit
for _, e := range entries {
listed[localKey(e.Module, e.LocalName)] = true
if e.Binding == nil {
if len(e.Settings) > 0 {
noBinding++
}
continue
}
byModule[e.Module] = append(byModule[e.Module], e)
@@ -384,7 +425,7 @@ func (v *BindingsView) take(p *Part, entries []meshbroker.Held) (written, rewrit
}
b, berr := p.bound(module)
if berr != nil {
return written, rewritten, noBinding, berr
return written, rewritten, fromFields, berr
}
changed := false
for _, e := range byModule[module] {
@@ -396,28 +437,35 @@ func (v *BindingsView) take(p *Part, entries []meshbroker.Held) (written, rewrit
v.logf("the binding of %s %s in the answer of %s does not read: not written", module, e.LocalName, CallBrokerList)
continue
}
cause := "the first view of the bindings (" + CallBrokerList + ")"
if composed[localKey(module, e.LocalName)] {
cause = "the first view of the bindings, composed from " + CallBrokerList + "'s fields"
}
was, had := b.Bindings[e.LocalName]
if had && sameBinding(was, bind) {
continue
}
b.Bindings[e.LocalName] = bind
changed = true
if composed[localKey(module, e.LocalName)] {
fromFields++
}
if had {
rewritten++
v.logf("the binding of %s %s was rewritten, the cause the first view of the bindings (%s)", module, e.LocalName, CallBrokerList)
v.logf("the binding of %s %s was rewritten, the cause %s", module, e.LocalName, cause)
} else {
written++
v.logf("the binding of %s %s was written, the cause the first view of the bindings (%s)", module, e.LocalName, CallBrokerList)
v.logf("the binding of %s %s was written, the cause %s", module, e.LocalName, cause)
}
}
if changed {
raw, _ := json.Marshal(b)
if _, perr := p.Store.Put(StateMemberships, BoundKey(module), raw); perr != nil {
return written, rewritten, noBinding, fmt.Errorf("its own %s %s cannot be written: %w", StateMemberships, BoundKey(module), perr)
return written, rewritten, fromFields, fmt.Errorf("its own %s %s cannot be written: %w", StateMemberships, BoundKey(module), perr)
}
}
}
return written, rewritten, noBinding, v.unlisted(p, listed, touched)
return written, rewritten, fromFields, v.unlisted(p, listed, touched)
}
// unlisted keeps every binding the answer does not list and raises its condition, one per binding; a binding it lists again
sdk/parts/receive.go
+8 -3
View File
@@ -239,8 +239,8 @@ func (r *Receiver[M]) arm(id string) error {
}
// Take takes one chunk of a part from the transfer's opener: chunks of a part arrive in sequence from 0, the last
// marked; the whole part must hash to its digest, else its chunks are dropped and the part is lacked again. Belongs,
// when given, refuses a transfer the caller's own shape says is not this call's (another seat's, say). When the last
// marked, and a chunk 0 again starts the part afresh (its earlier chunks are dropped); the whole part must hash to its
// digest, else its chunks are dropped and the part is lacked again. Belongs, when given, refuses a transfer the caller's own shape says is not this call's (another seat's, say). When the last
// lacked part arrives the transfer ends: the caller's check, then its store.
func (r *Receiver[M]) Take(id, opener, digest string, sequence int, last bool, data []byte, belongs func(Transfer[M]) error) (Received, error) {
t, found, err := r.Transfers.Get(id)
@@ -264,9 +264,14 @@ func (r *Receiver[M]) Take(id, opener, digest string, sequence int, last bool, d
return Received{}, fmt.Errorf("transfer %s names no part %s", id, digest)
case slices.Contains(t.Held, digest):
return Received{ID: id, Lacks: t.Lacks()}, nil
case sequence != t.Chunks[digest]:
case sequence != t.Chunks[digest] && sequence != 0:
return Received{}, fmt.Errorf("part %s of transfer %s waits for chunk %d, not %d", digest, id, t.Chunks[digest], sequence)
}
if sequence == 0 && t.Chunks[digest] != 0 {
// A part the holder lacks travels again whole from chunk 0 (design 51 Publishing): its earlier chunks are dropped.
r.dropChunks(t, digest)
t.Chunks[digest] = 0
}
if err := r.Objects.Put(r.chunkName(id, digest, sequence), data); err != nil {
return Received{}, fmt.Errorf("the objects did not take the chunk: %w", err)
}
sdk/parts/receive_test.go
+6 -1
View File
@@ -133,9 +133,14 @@ func aHolderTakesAFileInParts(t *testing.T) {
if _, err := pkg.Open("t1", "other", []Want{part}, packageShape{}); err == nil || !strings.Contains(err.Error(), "another caller") {
t.Errorf("reopen by another = %v", err)
}
if _, err := pkg.Take("t1", "builder", part.Digest, 0, false, blob[:1000], nil); err == nil || !strings.Contains(err.Error(), "waits for chunk 1") {
// 1966/AC203-2: a chunk that is neither the cursor nor 0 is refused, naming the cursor.
if _, err := pkg.Take("t1", "builder", part.Digest, 2, false, blob[:1000], nil); err == nil || !strings.Contains(err.Error(), "waits for chunk 1, not 2") {
t.Errorf("a chunk out of sequence = %v", err)
}
// 1966/AC203-1: a chunk 0 again (the part lacked travels whole again) drops the earlier chunks and takes it.
if _, err := pkg.Take("t1", "builder", part.Digest, 0, false, blob[:1000], nil); err != nil {
t.Errorf("chunk 0 again = %v", err)
}
a, err = pkg.Take("t1", "builder", part.Digest, 1, true, blob[1000:], nil)
if err != nil || !a.Ended || len(pkgLog) != 1 || pkgLog[0].outcome != "stored" || string(pkgLog[0].parts[part.Digest]) != string(blob) {
t.Fatalf("pkg end = %+v, %v, %+v", a, err, pkgLog)