From 3ec05dda1f41af7a4177403da37d99165c8408ad Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 10:48:08 +0100 Subject: [PATCH 1/5] assigner: a reads statement older than a reads-changed heard does not stand (issue 1975) The assigner keeps each reads-changed event it hears beside its statement (enginereads.Keeper.HeardChanged), reads a node as the event says while its statement is behind it, and asks again until the statement is as new. Builds: issue 1975 note 4 (the lead's handover; no design line) Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../cmd/mesh-delivery-assigner/main.go | 4 ++++ .../cmd/mesh-delivery-assigner/pass_test.go | 19 +++++++++++++++++++ 2 files changed, 23 insertions(+) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index 3ec09822..cb12c55a 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -569,6 +569,10 @@ func hearReadsChanged(ctx context.Context, js jetstream.JetStream, hub *process. logf("%s holds what is not %s, and %s is asked again all the same: %v", m.Subject(), enginereads.Event, node, err) reads.AskNode(node, enginereads.Event+" that does not read") default: + // What the event says is kept first: a statement older than it does not stand while the node is asked again. + if e, derr := enginereads.DecodeChanged(m.Data()); derr == nil { + reads.HeardChanged(e) + } reads.AskNode(node, enginereads.Event) } } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go index 5b077f78..e2a019c4 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go @@ -317,6 +317,25 @@ func TestAnAnswerCountsOnlyForTheAskingItAnswered(t *testing.T) { if asked["ace"] != 2 { t.Errorf("ace, back online, was asked %d times, want 2", asked["ace"]) } + // Issue 1975: ace's engine answered its bootstrapped seats, then stated the full list and said so; an answer older + // than that event does not stand: the rules read what the event says, and the node is asked until it is as new. + stale := true + bootstrapped := func(node, _ string, _ any) (json.RawMessage, error) { + if stale { + stale = false + return json.Marshal(nodeengine.ManifestReadsAnswer{V: 1, Node: node, ManifestReads: []string{"seat:a"}, StatedAt: "2026-11-12T01:14:36.498Z"}) + } + return json.Marshal(nodeengine.ManifestReadsAnswer{V: 1, Node: node, ManifestReads: []string{"seat:a", "seat:b"}, StatedAt: "2026-11-12T01:14:39.001Z"}) + } + rk := &enginereads.Keeper{Call: bootstrapped, Sync: true} + rk.AskNode("ace", "the assigner started") + rk.HeardChanged(nodeengine.ManifestReadsChangedEvent{V: 1, Node: "ace", Why: enginereads.ReadsStarted, At: "2026-11-12T01:14:39.002Z", ManifestReads: []string{"seat:a", "seat:b"}}) + rp := &passer{w: w, reads: rk, online: map[string]bool{"ace": true}} + rn := w.nodes(nil, nil) + rp.engineReads(rn) + if !slices.Equal(rn.Reads["ace"], []string{"seat:a", "seat:b"}) { + t.Errorf("the rules read the statement older than the event heard: %v", rn.Reads["ace"]) + } // ADR 0668 §3 (review round 1): a node whose engine answers nothing while its machine is heard is asked again at each // pass once its ask ended, and past five minutes is the finding call-unanswered, cleared once it answers; none is // written while its machine is away. From a3306361ee7b5391628b391b34ed7160bfbe0dd5 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 10:48:08 +0100 Subject: [PATCH 2/5] checker: the state compositions is bounded at 256 MiB and a day, and a tool drops a check's kept requests Builds: the lead's handover of 2026-11-12 (principle 7, principle 4; no design line) Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../cmd/mesh-delivery-checker-tools/main.go | 26 +++++- .../internal/check/engines_test.go | 31 ++++++- .../internal/check/requests.go | 87 +++++++++++++++++-- .../mesh-delivery-checker/manifest_test.go | 5 +- modules/mesh-delivery-checker/module.json | 5 +- 5 files changed, 141 insertions(+), 13 deletions(-) diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go index d32b387c..eebac0e6 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go @@ -378,8 +378,8 @@ func RequestTools(r check.StateRead) []stdio.Tool { Description: "The composition request a check sent to a node: the composer's whole input for that node (the node-engine's composition input, the case's " + "manifests and verdicts, the seat list), kept exactly as sent so a session reproduces the check's composition (the composer run by hand over the same " + "request gives the check's answer). Kept in the checker's state compositions, per check key (group.heads, the 12-digit heads hash) and node, " + - "0.7 to 0.8 MB each (kept compressed, in pieces that each fit the state's 256 KiB value), for 3 days (the state's ttl-seconds). With a check and a node, that request whole; with a check, " + - "the nodes it kept one for; with neither, every kept request, each with its size, digest and when it was kept. Read-only.", + "0.7 to 0.8 MB each (kept compressed, in pieces that each fit the state's 256 KiB value), for a day (the state's ttl-seconds), the state holding at most 256 MiB (max-mib). With a check and a node, that request whole; with a check, " + + "the nodes it kept one for; with neither, every kept request, each with its size, digest and when it was kept. Read-only; " + check.RequestsDropTool + " drops a check's.", Input: map[string]any{"type": "object", "properties": map[string]any{ "check": map[string]any{"type": "string", "description": "the check's key, ., as the verdict and composition name it"}, "node": map[string]any{"type": "string", "description": "the node the request was sent for, with a check"}}}, @@ -390,6 +390,26 @@ func RequestTools(r check.StateRead) []stdio.Tool { }}} } +// RequestDropTools is the tool that drops a check's kept composition requests by hand, which frees the state when it is +// full (tools-are-how-a-module-is-spoken-to): it records the hand act through the seat mesh-hand-acts first. +func RequestDropTools(s check.StateDrop, acts views.HandActs) []stdio.Tool { + return []stdio.Tool{{Name: check.RequestsDropTool, + Description: "Drop the composition requests the checker kept for one check (" + check.RequestsReadTool + " lists them): every node's, and the pieces " + + "no other kept request names. The hand act is recorded through the seat mesh-hand-acts first, and a drop whose act could not be recorded is not made. " + + "Answers the nodes and pieces dropped. Use it when a check's words say its request could not be kept in the state compositions because it is full; " + + "a request is dropped by the state itself a day after it was kept.", + Input: map[string]any{"type": "object", "required": []string{"check", "why"}, "properties": map[string]any{ + "check": map[string]any{"type": "string", "description": "the check's key, ., whose kept requests are dropped"}, + "why": map[string]any{"type": "string", "description": "why they are dropped, in words; recorded as the hand act"}}}, + Run: func(args map[string]any) (any, error) { + key, _ := args["check"].(string) + why, _ := args["why"].(string) + return check.DropRequests(s, key, why, func(what string) error { + return acts.RecordHandAct(views.HandAct{What: what, Why: why, Verb: check.RequestsDropTool}) + }) + }}} +} + // RecoveryTools are the tools that steer a check by hand (tools-are-how-a-module-is-spoken-to): ask a check again whose // wait judged nothing (the lost check of 2026-10-28). They write the checker's verdicts through w. func RecoveryTools(r check.Reader, w check.VerdictEntry) []stdio.Tool { @@ -469,6 +489,8 @@ func main() { tools = append(tools, views.StatusTools(check.CheckerModule, "checker_first_views", "checker_first_view_again", stdio.State(views.StatusState), meshhandactsclient.Recorder{Ask: stdio.Ask}, time.Now)...) tools = append(tools, SeatInputTools(r)...) tools = append(tools, RequestTools(read(stdio.State(check.RequestsStateName)))...) + requests := stdio.State(check.RequestsStateName) + tools = append(tools, RequestDropTools(check.StateDrop{StateRead: read(requests), Delete: requests.Delete}, meshhandactsclient.Recorder{Ask: stdio.Ask})...) if err := stdio.Serve("", logged(tools)); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-checker-tools] %v\n", err) os.Exit(1) diff --git a/modules/mesh-delivery-checker/internal/check/engines_test.go b/modules/mesh-delivery-checker/internal/check/engines_test.go index 1c76571e..36a96514 100644 --- a/modules/mesh-delivery-checker/internal/check/engines_test.go +++ b/modules/mesh-delivery-checker/internal/check/engines_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "mesh-delivery-checker/rules" "strings" "sync" @@ -235,8 +236,34 @@ func TestTheCompositionIsMadeFromTheEnginesAnswers(t *testing.T) { } store.err = errors.New("the state is full") raw, _ = ec.Compose(context.Background(), host) - if v := rules.JudgeEngines(raw); v.State != meshdeliverycheckerclient.CompositionFail || !strings.Contains(v.Words, "could not be kept") || !strings.Contains(v.Words, "the state is full") { - t.Errorf("a request that cannot be kept gave %+v", v) + if v := rules.JudgeEngines(raw); v.State != meshdeliverycheckerclient.CompositionFail || !strings.Contains(v.Words, "could not be kept") || !strings.Contains(v.Words, "the state is full") || !strings.Contains(v.Words, RequestsDropTool) { + t.Errorf("a request that cannot be kept gave %+v, which must name %s", v, RequestsDropTool) + } + // The drop tool: records the hand act first (a drop whose act is not recorded is not made), then drops that check's + // requests and the pieces no other kept request names; another check's stay. + store.err = nil + other := RequestKeeper{Store: store} + if err := other.Keep(context.Background(), "g.other", "anchor", []byte("another check's request")); err != nil { + t.Fatal(err) + } + drop := StateDrop{StateRead: store.read(), Delete: func(k string) error { store.mu.Lock(); delete(store.vals, k); store.mu.Unlock(); return nil }} + before := len(store.vals) + if _, err := DropRequests(drop, keyOfHost, "the state is full", func(string) error { return errors.New("no holder") }); err == nil || len(store.vals) != before { + t.Errorf("a drop whose hand act was not recorded was made: %v, %d of %d kept", err, len(store.vals), before) + } + var acts []string + answer, err := DropRequests(drop, keyOfHost, "the state is full", func(what string) error { acts = append(acts, what); return nil }) + if b, _ := json.Marshal(answer); err != nil || len(acts) != 1 || !strings.Contains(string(b), `"dropped_requests":["anchor"]`) { + t.Errorf("the drop answered %s, %v, acts %v", b, err, acts) + } + if _, still := store.vals[RequestKey(keyOfHost, "anchor")]; still { + t.Errorf("the dropped check's request is kept still") + } + if got, err := ReadRequests(store.read(), "g.other", "anchor"); err != nil || !strings.Contains(fmt.Sprint(got), "another check's request") { + t.Errorf("another check's request after the drop: %v, %v", got, err) + } + if _, err := DropRequests(drop, keyOfHost, "again", func(string) error { return nil }); err == nil { + t.Errorf("a check with no kept request was dropped") } ec.Requests = nil // Issue 1921: the composer is built at the group's heads, also a member that publishes no package yet. diff --git a/modules/mesh-delivery-checker/internal/check/requests.go b/modules/mesh-delivery-checker/internal/check/requests.go index 46d4531d..64838a3c 100644 --- a/modules/mesh-delivery-checker/internal/check/requests.go +++ b/modules/mesh-delivery-checker/internal/check/requests.go @@ -26,8 +26,9 @@ import ( // of the check and node holds a small document naming the request's digest, size and number of pieces, and each piece is // a value of its own under RequestChunkKey, of at most RequestChunkBytes. The kept data is the checker's, a foundation // part's, so it stays in the old section `state` (ADR 0619 §2) and is read through the tool runner's state read, which -// serves no object store of the old sections. The manifest's ttl-seconds is the lifetime (3 days) of the pointer and of -// every piece: the bus drops a value past it, so nothing waits on a timer here. A check kept again overwrites the key; +// serves no object store of the old sections. The manifest's ttl-seconds is the lifetime (1 day) of the pointer and of +// every piece and its max-mib the most the state holds (256): the bus drops a value past the one and refuses a put past the +// other, so nothing waits on a timer here; a state that is full says so in the check's words, which name RequestsDropTool. A check kept again overwrites the key; // a request whose pieces cannot all be put is refused naming the size (never dropped, never half kept: the pointer is // put after its pieces). const ( @@ -36,6 +37,8 @@ const ( RequestsBucket = CheckerModule + "_" + RequestsStateName // RequestsReadTool is the tool that reads a kept request. RequestsReadTool = "checker_composition_requests" + // RequestsDropTool is the tool that drops a check's kept requests by hand (a state that is full is freed with it). + RequestsDropTool = "checker_composition_requests_drop" // RequestChunkBytes is the most compressed bytes one piece holds: base64 of it, in its document, stays well under // the state's 256 KiB value. RequestChunkBytes = 128 << 10 @@ -110,8 +113,8 @@ func (k RequestKeeper) Keep(ctx context.Context, key, node string, body []byte) digest := hex.EncodeToString(sum[:]) pieces := (packed.Len() + RequestChunkBytes - 1) / RequestChunkBytes fail := func(err error) error { - return fmt.Errorf("the composition request of check %s to %s (%d bytes, %d compressed in %d pieces) could not be kept in the state %s: %w", - key, node, len(body), packed.Len(), pieces, RequestsStateName, err) + return fmt.Errorf("the composition request of check %s to %s (%d bytes, %d compressed in %d pieces) could not be kept in the state %s: %w (when the state is full, free it by dropping the kept requests of a check no longer read, with the tool %s)", + key, node, len(body), packed.Len(), pieces, RequestsStateName, err, RequestsDropTool) } for i := 0; i < pieces; i++ { end := min((i+1)*RequestChunkBytes, packed.Len()) @@ -190,7 +193,7 @@ func unpackRequest(s StateRead, raw []byte) (KeptRequest, []byte, error) { return kept, nil, fmt.Errorf("is no kept request: %w", err) } if kept.V != 2 { - return kept, nil, fmt.Errorf("is a kept request of version %d, which this reader does not read (it reads version 2; an older one is past its lifetime within 3 days)", kept.V) + return kept, nil, fmt.Errorf("is a kept request of version %d, which this reader does not read (it reads version 2; an older one is past its lifetime within a day)", kept.V) } var packed []byte for i := 0; i < kept.Pieces; i++ { @@ -221,3 +224,77 @@ func unpackRequest(s StateRead, raw []byte) (KeptRequest, []byte, error) { } return kept, body, nil } + +// StateDrop is the state of kept requests as the drop tool needs it: its keys, a value, and a key removed. +type StateDrop struct { + StateRead + Delete func(key string) error +} + +// DropRequests drops the requests kept for a check, after record (the hand act, recorded first: a drop that is not +// recorded is not made), and answers what it dropped: the nodes, and the pieces. A piece another kept request still names +// (the same request kept for another check has the same digest) stays. A check naming no kept request is refused. +func DropRequests(s StateDrop, key, why string, record func(what string) error) (any, error) { + if strings.TrimSpace(key) == "" || strings.TrimSpace(why) == "" { + return nil, fmt.Errorf("drop takes check (its key, group.heads) and why") + } + if s.Keys == nil || s.Get == nil || s.Delete == nil { + return nil, fmt.Errorf("the state %s cannot be changed here", RequestsStateName) + } + keys, err := s.Keys() + if err != nil { + return nil, err + } + sort.Strings(keys) + var drop []string + kept := map[string]bool{} // digests the requests that stay name + dropping := map[string]bool{} + for _, k := range keys { + if isPieceKey(k) { + continue + } + raw, err := s.Get(k) + if err != nil { + return nil, err + } + var r KeptRequest + if len(raw) == 0 || json.Unmarshal(raw, &r) != nil { + continue + } + if strings.HasPrefix(k, key+"/") { + drop = append(drop, k) + dropping[r.Digest] = true + continue + } + kept[r.Digest] = true + } + if len(drop) == 0 { + return nil, fmt.Errorf("no composition request of check %s is kept (none was sent, or it is past its lifetime): nothing was dropped", key) + } + if err := record(fmt.Sprintf("dropped the %d kept composition requests of check %s (%s)", len(drop), key, RequestsStateName)); err != nil { + return nil, fmt.Errorf("the hand act was not recorded, so nothing was dropped: %w", err) + } + var pieces int + for _, k := range drop { + if err := s.Delete(k); err != nil { + return nil, fmt.Errorf("dropping %s: %w", k, err) + } + } + for _, k := range keys { + if !isPieceKey(k) { + continue + } + digest := strings.Split(strings.TrimPrefix(k, "pieces/"), "/")[0] + if dropping[digest] && !kept[digest] { + if err := s.Delete(k); err != nil { + return nil, fmt.Errorf("dropping %s: %w", k, err) + } + pieces++ + } + } + nodes := make([]string, len(drop)) + for i, k := range drop { + nodes[i] = strings.TrimPrefix(k, key+"/") + } + return map[string]any{"check": key, "dropped_requests": nodes, "dropped_pieces": pieces, "why": why}, nil +} diff --git a/modules/mesh-delivery-checker/manifest_test.go b/modules/mesh-delivery-checker/manifest_test.go index ac47d009..5451f26f 100644 --- a/modules/mesh-delivery-checker/manifest_test.go +++ b/modules/mesh-delivery-checker/manifest_test.go @@ -23,6 +23,7 @@ type state struct { Name string `json:"name"` History int `json:"history,omitempty"` TTL int `json:"ttl-seconds,omitempty"` + MaxMiB int `json:"max-mib,omitempty"` } type artifact struct { @@ -108,7 +109,7 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1 t.Errorf("module = %q", m.Module) } // The histories of the checker's four seat states (checked against internal/check names below); the names follow the seat mesh-delivery-checker. - if want := []state{{Name: "asks", History: 1}, {Name: "hearing", History: 1}, {Name: "inputs", TTL: 86400}, {Name: "compositions", TTL: 259200}, {Name: "kept", History: 1}, {Name: "owed", History: 5}, {Name: "previews", History: 5}, {Name: "tries", History: 1}, {Name: "verdicts", History: 5}, {Name: "first-views", History: 1}, {Name: "findings", History: 1}}; !reflect.DeepEqual(m.State, want) { + if want := []state{{Name: "asks", History: 1}, {Name: "hearing", History: 1}, {Name: "inputs", TTL: 86400}, {Name: "compositions", TTL: 86400, MaxMiB: 256}, {Name: "kept", History: 1}, {Name: "owed", History: 5}, {Name: "previews", History: 5}, {Name: "tries", History: 1}, {Name: "verdicts", History: 5}, {Name: "first-views", History: 1}, {Name: "findings", History: 1}}; !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-node.steps", "mesh-build-node.toolchains", @@ -186,7 +187,7 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1 // The read tools (issue 1686) are served by the tools bundle, not by the checker's process. // The try's tools (ADR 0501 §5; the end by hand, issue 734) beside them: mesh-controller's try calls them. A lost check // is asked again by hand (checker_ask_again). - if want := []string{"checker_ask_again", "checker_composition_requests", "checker_first_view_again", "checker_first_views", "checker_inputs", "checker_kept", "checker_previews", "checker_pull", "checker_status", "checker_tries", "checker_try", "checker_try_end", "checker_try_read", "checker_verdicts"}; !reflect.DeepEqual(sorted(m.Tools), want) { + if want := []string{"checker_ask_again", "checker_composition_requests", "checker_composition_requests_drop", "checker_first_view_again", "checker_first_views", "checker_inputs", "checker_kept", "checker_previews", "checker_pull", "checker_status", "checker_tries", "checker_try", "checker_try_end", "checker_try_read", "checker_verdicts"}; !reflect.DeepEqual(sorted(m.Tools), want) { t.Errorf("tools = %v, want %v", m.Tools, want) } // The try is the checker's own seat's verbs (ADR 0547 §10, issue 774), served beside the tools by the same code; so is diff --git a/modules/mesh-delivery-checker/module.json b/modules/mesh-delivery-checker/module.json index 49efcfe4..07d368dd 100644 --- a/modules/mesh-delivery-checker/module.json +++ b/modules/mesh-delivery-checker/module.json @@ -1,6 +1,6 @@ { "module": "mesh-delivery-checker", - "version": "1", + "version": "2", "slug": "checker", "claims": [ { @@ -37,7 +37,7 @@ { "name": "asks", "history": 1 }, { "name": "hearing", "history": 1 }, { "name": "inputs", "ttl-seconds": 86400 }, - { "name": "compositions", "ttl-seconds": 259200 }, + { "name": "compositions", "ttl-seconds": 86400, "max-mib": 256 }, { "name": "kept", "history": 1 }, { "name": "owed", "history": 5 }, { "name": "previews", "history": 5 }, @@ -64,6 +64,7 @@ "tools": [ "checker_ask_again", "checker_composition_requests", + "checker_composition_requests_drop", "checker_inputs", "checker_kept", "checker_previews", From e12f4b608fc62b6bbf0dcada299841d374ad774c Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 11:00:33 +0100 Subject: [PATCH 3/5] assigner: the test drives the reads-changed handler (issue 1975 note 6) Builds: issue 1975 note 6 (the lead's review round 1; no design line) Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../cmd/mesh-delivery-assigner/main.go | 20 ++++++++++------ .../cmd/mesh-delivery-assigner/pass_test.go | 23 +++++++++++++------ 2 files changed, 29 insertions(+), 14 deletions(-) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index cb12c55a..eb8d2f19 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -552,13 +552,10 @@ func sayVia(ctx context.Context, js jetstream.JetStream) func(oc.Condition) erro } } -// hearReadsChanged hears the node-engine seat's event manifest-reads-changed on the assigner's one hub of its durable -// consumer on EVENTS, and asks the node it names again; the hub also hands the indexer's views to the followers of -// indexerViews, and starts once both are registered. -func hearReadsChanged(ctx context.Context, js jetstream.JetStream, hub *process.EventHub, reads *enginereads.Keeper, findings assign.Findings, live func(), logf func(string, ...any)) { - account := process.Account{Node: thisNode, Module: assignment.Module} - w := notHeardWriter(findings, logf) - handle := func(m jetstream.Msg) { +// readsChangedHandler is what the assigner does with each event manifest-reads-changed: acknowledges it, keeps what it says +// of the node (a statement older than it does not stand) and asks the node again. +func readsChangedHandler(reads *enginereads.Keeper, logf func(string, ...any)) func(jetstream.Msg) { + return func(m jetstream.Msg) { if err := m.Ack(); err != nil { logf("acknowledging %s: %v", m.Subject(), err) } @@ -576,6 +573,15 @@ func hearReadsChanged(ctx context.Context, js jetstream.JetStream, hub *process. reads.AskNode(node, enginereads.Event) } } +} + +// hearReadsChanged hears the node-engine seat's event manifest-reads-changed on the assigner's one hub of its durable +// consumer on EVENTS, and asks the node it names again; the hub also hands the indexer's views to the followers of +// indexerViews, and starts once both are registered. +func hearReadsChanged(ctx context.Context, js jetstream.JetStream, hub *process.EventHub, reads *enginereads.Keeper, findings assign.Findings, live func(), logf func(string, ...any)) { + account := process.Account{Node: thisNode, Module: assignment.Module} + w := notHeardWriter(findings, logf) + handle := readsChangedHandler(reads, logf) err := hub.Handle(enginereads.SubjectPrefix+"*", handle) // At each bind every node the assigner holds a statement of is asked again: an event said before its consumer existed // is not delivered to it (issue 1966 note 155: ace's bootstrapped statement stood; ADR 0668 §4). diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go index e2a019c4..570aca3e 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go @@ -319,17 +319,15 @@ func TestAnAnswerCountsOnlyForTheAskingItAnswered(t *testing.T) { } // Issue 1975: ace's engine answered its bootstrapped seats, then stated the full list and said so; an answer older // than that event does not stand: the rules read what the event says, and the node is asked until it is as new. - stale := true + // The engine's verb keeps answering the bootstrapped statement (its stated-at is older than the event); the event is + // handed to the assigner's own handler (readsChangedHandler, what hearReadsChanged registers) as the bus delivers it. bootstrapped := func(node, _ string, _ any) (json.RawMessage, error) { - if stale { - stale = false - return json.Marshal(nodeengine.ManifestReadsAnswer{V: 1, Node: node, ManifestReads: []string{"seat:a"}, StatedAt: "2026-11-12T01:14:36.498Z"}) - } - return json.Marshal(nodeengine.ManifestReadsAnswer{V: 1, Node: node, ManifestReads: []string{"seat:a", "seat:b"}, StatedAt: "2026-11-12T01:14:39.001Z"}) + return json.Marshal(nodeengine.ManifestReadsAnswer{V: 1, Node: node, ManifestReads: []string{"seat:a"}, StatedAt: "2026-11-12T01:14:36.498Z"}) } rk := &enginereads.Keeper{Call: bootstrapped, Sync: true} rk.AskNode("ace", "the assigner started") - rk.HeardChanged(nodeengine.ManifestReadsChangedEvent{V: 1, Node: "ace", Why: enginereads.ReadsStarted, At: "2026-11-12T01:14:39.002Z", ManifestReads: []string{"seat:a", "seat:b"}}) + body, _ := json.Marshal(nodeengine.ManifestReadsChangedEvent{V: 1, Node: "ace", Why: enginereads.ReadsStarted, At: "2026-11-12T01:14:39.002Z", ManifestReads: []string{"seat:a", "seat:b"}}) + readsChangedHandler(rk, t.Logf)(readsMsg{subject: enginereads.SubjectPrefix + "ace", data: body}) rp := &passer{w: w, reads: rk, online: map[string]bool{"ace": true}} rn := w.nodes(nil, nil) rp.engineReads(rn) @@ -1510,3 +1508,14 @@ func readAskID(t *testing.T, p *passer, key string) string { } return d.Ask.ID } + +// readsMsg is a message of the events stream as the assigner's handler reads it. +type readsMsg struct { + jetstream.Msg + subject string + data []byte +} + +func (m readsMsg) Ack() error { return nil } +func (m readsMsg) Subject() string { return m.subject } +func (m readsMsg) Data() []byte { return m.data } From f28c49a9766211fda000508be048aa40316f48e6 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 11:00:33 +0100 Subject: [PATCH 4/5] grants: a state's declared max-mib reaches its bucket, and a held bucket is raised to it (issue 1975 note 6) Builds: issue 1975 note 6 (the lead's review round 1; no design line) Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../mesh-grants/internal/buscompose/create.go | 6 ++- .../internal/buscompose/create_test.go | 2 + .../mesh-grants/internal/buscompose/state.go | 2 + .../mesh-grants/internal/grants/assertions.go | 7 +++- .../mesh-grants/internal/grants/records.go | 2 +- .../internal/grants/statecap_test.go | 40 +++++++++++++++++++ 6 files changed, 56 insertions(+), 3 deletions(-) create mode 100644 modules/mesh-grants/internal/grants/statecap_test.go diff --git a/modules/mesh-grants/internal/buscompose/create.go b/modules/mesh-grants/internal/buscompose/create.go index f4d76a68..5c534ed6 100644 --- a/modules/mesh-grants/internal/buscompose/create.go +++ b/modules/mesh-grants/internal/buscompose/create.go @@ -25,12 +25,16 @@ func BucketStreamConfig(b Bucket) jetstream.StreamConfig { if age > 0 && age < dupes { dupes = age } + maxBytes := int64(StateMaxBytes) + if b.MaxMiB > 0 { + maxBytes = int64(b.MaxMiB) << 20 + } cfg := jetstream.StreamConfig{ Name: "KV_" + b.Bucket(), Description: b.Why(), Subjects: []string{"$KV." + b.Bucket() + ".>"}, MaxMsgsPerSubject: int64(history), - MaxBytes: StateMaxBytes, + MaxBytes: maxBytes, MaxAge: age, MaxMsgSize: StateMaxValueBytes, Storage: jetstream.FileStorage, diff --git a/modules/mesh-grants/internal/buscompose/create_test.go b/modules/mesh-grants/internal/buscompose/create_test.go index 0df9580e..64f8bc47 100644 --- a/modules/mesh-grants/internal/buscompose/create_test.go +++ b/modules/mesh-grants/internal/buscompose/create_test.go @@ -10,6 +10,7 @@ import ( // Defends issue 1790 (ADR 0547 §5): the grants part creates a declared bucket and stream as the controller asserts them. func declaredStreamsAreCreatedAsTheControllerAssertsThem(t *testing.T) { plain := BucketStreamConfig(Bucket{Module: "mesh-vault", Name: "bus-accounts"}) + declared := BucketStreamConfig(Bucket{Module: "c", Name: "compositions", MaxMiB: 256}) keyed := BucketStreamConfig(Bucket{Module: "m", Name: "presence", History: 5, TTLSeconds: 60, KeyTTL: true}) queue := ModuleStreamConfig(ModuleStream{Module: "b", Name: "ASKS", Events: []string{"ask"}, MaxAgeDays: 7, MaxMiB: 32, WorkQueue: true, Duplicates: 1}) @@ -21,6 +22,7 @@ func declaredStreamsAreCreatedAsTheControllerAssertsThem(t *testing.T) { {"bucket subjects", plain.Subjects[0], "$KV.mesh-vault_bus-accounts.>"}, {"bucket history defaults to one", plain.MaxMsgsPerSubject, int64(1)}, {"bucket cap", plain.MaxBytes, int64(StateMaxBytes)}, + {"bucket cap as its state declares it (max-mib)", declared.MaxBytes, int64(256 << 20)}, {"bucket value cap", plain.MaxMsgSize, int32(StateMaxValueBytes)}, {"bucket duplicate window", plain.Duplicates, 2 * time.Minute}, {"bucket discards new", plain.Discard, jetstream.DiscardNew}, diff --git a/modules/mesh-grants/internal/buscompose/state.go b/modules/mesh-grants/internal/buscompose/state.go index 7a57694a..072b239e 100644 --- a/modules/mesh-grants/internal/buscompose/state.go +++ b/modules/mesh-grants/internal/buscompose/state.go @@ -36,6 +36,8 @@ type Bucket struct { History int // TTLSeconds is how long a value lives; zero is until replaced or deleted. TTLSeconds int + // MaxMiB is the bucket's cap in MiB as its state declares (max-mib); zero is StateMaxBytes. + MaxMiB int // PerMachine says each machine's instance writes only the key named for its machine (novox/hq ADR // 0260), and is granted that key and no other. PerMachine bool diff --git a/modules/mesh-grants/internal/grants/assertions.go b/modules/mesh-grants/internal/grants/assertions.go index 03f1c8bc..32b15ed3 100644 --- a/modules/mesh-grants/internal/grants/assertions.go +++ b/modules/mesh-grants/internal/grants/assertions.go @@ -277,9 +277,14 @@ func Composed(m catalogue.Manifest, nodes []string) []Item { if per < 1 { per = 1 } + // A bucket's size cap is composed only where its state declares one (max-mib); the default is the bus's own. + var maxBytes int64 + if b.MaxMiB > 0 { + maxBytes = buscompose.BucketStreamConfig(b).MaxBytes + } out = append(out, Item{Kind: "bucket", Stream: "KV_" + b.Bucket(), Composed: Limits{ Subjects: []string{"$KV." + b.Bucket() + ".>"}, Retention: string(buscompose.RetentionLimits), - MaxAgeHours: b.TTLSeconds / 3600, MaxPerSubject: per}}) + MaxAgeHours: b.TTLSeconds / 3600, MaxPerSubject: per, MaxBytes: maxBytes}}) } for _, l := range logsOf(m) { c := l.Config() diff --git a/modules/mesh-grants/internal/grants/records.go b/modules/mesh-grants/internal/grants/records.go index ae8842d7..b4a78958 100644 --- a/modules/mesh-grants/internal/grants/records.go +++ b/modules/mesh-grants/internal/grants/records.go @@ -269,7 +269,7 @@ func issuedTools(m catalogue.Manifest) []string { func bucketsOf(m catalogue.Manifest) []buscompose.Bucket { var out []buscompose.Bucket for _, s := range m.State { - out = append(out, buscompose.Bucket{Module: m.Module, Name: s.Name, History: s.History, TTLSeconds: s.TTLSeconds, + out = append(out, buscompose.Bucket{Module: m.Module, Name: s.Name, History: s.History, TTLSeconds: s.TTLSeconds, MaxMiB: s.MaxMiB, PerMachine: s.PerMachine, KeyTTL: s.KeyTTL, ReadOnly: s.ReadOnly}) } return out diff --git a/modules/mesh-grants/internal/grants/statecap_test.go b/modules/mesh-grants/internal/grants/statecap_test.go new file mode 100644 index 00000000..3289b718 --- /dev/null +++ b/modules/mesh-grants/internal/grants/statecap_test.go @@ -0,0 +1,40 @@ +package grants + +import ( + "testing" + + "git.novox.be/novox/mesh-sdk/go/catalogue" +) + +// raiser is a Creator that holds one stream at a cap and records the cap it is raised to. +type raiser struct { + Creator + to int64 +} + +func (r *raiser) RaiseMaxBytes(_ string, to int64) (int64, bool, error) { + r.to = to + return 64 << 20, true, nil +} + +// A state's declared max-mib reaches the bucket (issue 1975 note 6): the composed stream carries it, and a bucket the bus +// holds at the default 64 MiB is raised to it by the assertion. +func TestAStatesDeclaredCapReachesItsBucket(t *testing.T) { + m := catalogue.Manifest{Module: "mesh-delivery-checker", State: []catalogue.StateDeclaration{{Name: "compositions", MaxMiB: 256}}} + var composed *Item + for _, it := range Composed(m, nil) { + if it.Stream == "KV_mesh-delivery-checker_compositions" { + composed = &it + } + } + if composed == nil || composed.Composed.MaxBytes != 256<<20 { + t.Fatalf("the composed bucket is %+v, want a cap of 256 MiB", composed) + } + a := Assertion{Manifest: "d", Items: []Item{*composed}} + a.Items[0].Held = &Limits{MaxBytes: 64 << 20} + r := &raiser{} + _, raises := createMissing(r, &creations{}, &a, m, nil, false) + if r.to != 256<<20 || len(raises) != 1 { + t.Errorf("the held bucket was raised to %d (%v), want 256 MiB", r.to, raises) + } +} From b83c546831b0392fe33aca84f64dfcb16f787f16 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 11:00:33 +0100 Subject: [PATCH 5/5] checker: compositions keeps ttl-seconds 259200 with max-mib 256; the texts say only what holds (issue 1975 note 6) Builds: issue 1975 note 6 (the lead's review round 1; no design line) Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../cmd/mesh-delivery-checker-tools/main.go | 4 ++-- modules/mesh-delivery-checker/internal/check/requests.go | 4 ++-- modules/mesh-delivery-checker/manifest_test.go | 2 +- modules/mesh-delivery-checker/module.json | 2 +- 4 files changed, 6 insertions(+), 6 deletions(-) diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go index eebac0e6..28504499 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go @@ -378,7 +378,7 @@ func RequestTools(r check.StateRead) []stdio.Tool { Description: "The composition request a check sent to a node: the composer's whole input for that node (the node-engine's composition input, the case's " + "manifests and verdicts, the seat list), kept exactly as sent so a session reproduces the check's composition (the composer run by hand over the same " + "request gives the check's answer). Kept in the checker's state compositions, per check key (group.heads, the 12-digit heads hash) and node, " + - "0.7 to 0.8 MB each (kept compressed, in pieces that each fit the state's 256 KiB value), for a day (the state's ttl-seconds), the state holding at most 256 MiB (max-mib). With a check and a node, that request whole; with a check, " + + "0.7 to 0.8 MB each (kept compressed, in pieces that each fit the state's 256 KiB value), for 3 days (the state's ttl-seconds), the state holding at most 256 MiB (max-mib, which mesh-grants raises a held bucket to). With a check and a node, that request whole; with a check, " + "the nodes it kept one for; with neither, every kept request, each with its size, digest and when it was kept. Read-only; " + check.RequestsDropTool + " drops a check's.", Input: map[string]any{"type": "object", "properties": map[string]any{ "check": map[string]any{"type": "string", "description": "the check's key, ., as the verdict and composition name it"}, @@ -397,7 +397,7 @@ func RequestDropTools(s check.StateDrop, acts views.HandActs) []stdio.Tool { Description: "Drop the composition requests the checker kept for one check (" + check.RequestsReadTool + " lists them): every node's, and the pieces " + "no other kept request names. The hand act is recorded through the seat mesh-hand-acts first, and a drop whose act could not be recorded is not made. " + "Answers the nodes and pieces dropped. Use it when a check's words say its request could not be kept in the state compositions because it is full; " + - "a request is dropped by the state itself a day after it was kept.", + "the state itself drops a request 3 days after it was kept.", Input: map[string]any{"type": "object", "required": []string{"check", "why"}, "properties": map[string]any{ "check": map[string]any{"type": "string", "description": "the check's key, ., whose kept requests are dropped"}, "why": map[string]any{"type": "string", "description": "why they are dropped, in words; recorded as the hand act"}}}, diff --git a/modules/mesh-delivery-checker/internal/check/requests.go b/modules/mesh-delivery-checker/internal/check/requests.go index 64838a3c..0345ec80 100644 --- a/modules/mesh-delivery-checker/internal/check/requests.go +++ b/modules/mesh-delivery-checker/internal/check/requests.go @@ -26,7 +26,7 @@ import ( // of the check and node holds a small document naming the request's digest, size and number of pieces, and each piece is // a value of its own under RequestChunkKey, of at most RequestChunkBytes. The kept data is the checker's, a foundation // part's, so it stays in the old section `state` (ADR 0619 §2) and is read through the tool runner's state read, which -// serves no object store of the old sections. The manifest's ttl-seconds is the lifetime (1 day) of the pointer and of +// serves no object store of the old sections. The manifest's ttl-seconds is the lifetime (3 days) of the pointer and of // every piece and its max-mib the most the state holds (256): the bus drops a value past the one and refuses a put past the // other, so nothing waits on a timer here; a state that is full says so in the check's words, which name RequestsDropTool. A check kept again overwrites the key; // a request whose pieces cannot all be put is refused naming the size (never dropped, never half kept: the pointer is @@ -193,7 +193,7 @@ func unpackRequest(s StateRead, raw []byte) (KeptRequest, []byte, error) { return kept, nil, fmt.Errorf("is no kept request: %w", err) } if kept.V != 2 { - return kept, nil, fmt.Errorf("is a kept request of version %d, which this reader does not read (it reads version 2; an older one is past its lifetime within a day)", kept.V) + return kept, nil, fmt.Errorf("is a kept request of version %d, which this reader does not read (it reads version 2; an older one is past its lifetime within 3 days)", kept.V) } var packed []byte for i := 0; i < kept.Pieces; i++ { diff --git a/modules/mesh-delivery-checker/manifest_test.go b/modules/mesh-delivery-checker/manifest_test.go index 5451f26f..796113f8 100644 --- a/modules/mesh-delivery-checker/manifest_test.go +++ b/modules/mesh-delivery-checker/manifest_test.go @@ -109,7 +109,7 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1 t.Errorf("module = %q", m.Module) } // The histories of the checker's four seat states (checked against internal/check names below); the names follow the seat mesh-delivery-checker. - if want := []state{{Name: "asks", History: 1}, {Name: "hearing", History: 1}, {Name: "inputs", TTL: 86400}, {Name: "compositions", TTL: 86400, MaxMiB: 256}, {Name: "kept", History: 1}, {Name: "owed", History: 5}, {Name: "previews", History: 5}, {Name: "tries", History: 1}, {Name: "verdicts", History: 5}, {Name: "first-views", History: 1}, {Name: "findings", History: 1}}; !reflect.DeepEqual(m.State, want) { + if want := []state{{Name: "asks", History: 1}, {Name: "hearing", History: 1}, {Name: "inputs", TTL: 86400}, {Name: "compositions", TTL: 259200, MaxMiB: 256}, {Name: "kept", History: 1}, {Name: "owed", History: 5}, {Name: "previews", History: 5}, {Name: "tries", History: 1}, {Name: "verdicts", History: 5}, {Name: "first-views", History: 1}, {Name: "findings", History: 1}}; !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-node.steps", "mesh-build-node.toolchains", diff --git a/modules/mesh-delivery-checker/module.json b/modules/mesh-delivery-checker/module.json index 07d368dd..96a896b3 100644 --- a/modules/mesh-delivery-checker/module.json +++ b/modules/mesh-delivery-checker/module.json @@ -37,7 +37,7 @@ { "name": "asks", "history": 1 }, { "name": "hearing", "history": 1 }, { "name": "inputs", "ttl-seconds": 86400 }, - { "name": "compositions", "ttl-seconds": 86400, "max-mib": 256 }, + { "name": "compositions", "ttl-seconds": 259200, "max-mib": 256 }, { "name": "kept", "history": 1 }, { "name": "owed", "history": 5 }, { "name": "previews", "history": 5 },