diff --git a/modules/mesh-config/cmd/mesh-config/main.go b/modules/mesh-config/cmd/mesh-config/main.go index 9d7e424bc..09f0b32dd 100644 --- a/modules/mesh-config/cmd/mesh-config/main.go +++ b/modules/mesh-config/cmd/mesh-config/main.go @@ -251,17 +251,17 @@ func hearing(c *Config) { time.Sleep(wait) } }() - for _, pattern := range []string{"gitea.pull.merged", "gitea.pull.closed", "gitea.forge.pull-request.closed"} { + for _, pattern := range []string{"git.pull-merged", "git.pull-closed"} { go func(pattern string) { <-stdio.Served() // subscribed as soon as the runtime serves, never after a guessed wait: an event sent meanwhile is heard for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { err := stdio.Subscribe(pattern, func(e stdio.Envelope) error { - var p PullEvent - if json.Unmarshal(e.Body, &p) != nil { + p, ok := pullEventOf(e.Body) + if !ok { logf("[mesh-config] %s is not readable", pattern) return nil } - for _, s := range c.HeardPull(p, pattern == "gitea.pull.merged") { + for _, s := range c.HeardPull(p, pattern == "git.pull-merged") { logf("[mesh-config] %s", s) } return nil diff --git a/modules/mesh-config/cmd/mesh-config/pullforge_test.go b/modules/mesh-config/cmd/mesh-config/pullforge_test.go index e3d7fab70..e2418e65c 100644 --- a/modules/mesh-config/cmd/mesh-config/pullforge_test.go +++ b/modules/mesh-config/cmd/mesh-config/pullforge_test.go @@ -2,34 +2,53 @@ package main import ( "encoding/json" - "os" "testing" ) -// toolForge.Pull reads gitea_get_pull_request as the forge answers it (testdata/captured), bare or in the wrappings the -// runtime puts around a tool's answer. -func TestToolForgePullReadsTheCapturedAnswers(t *testing.T) { +// toolForge.Pull reads the seat git's answer of the verb pull (design 47 G §654), bare or in the wrappings the runtime puts +// around an answer; pullEventOf reads the seat's pull-merged and pull-closed bodies. +func TestToolForgePullReadsTheSeatGitAnswersAndEventsAreRead(t *testing.T) { + var asked struct { + key string + body map[string]any + } for _, tc := range []struct { - file string - number int + answer string open, merged bool head string }{ - {"get_pull_request_closed_merged_435.json", 435, false, true, "7b5f7f84acb73aaf8189200548ff69fd9aac04b2"}, - {"get_pull_request_open_434.json", 434, true, false, "c948f3a61c74257e6e3147effb396d9d6434f491"}, + {`{"pull":{"repository":"novox/mesh-catalog","number":435,"state":"merged","head":"7b5f7f84acb73aaf8189200548ff69fd9aac04b2"}}`, false, true, "7b5f7f84acb73aaf8189200548ff69fd9aac04b2"}, + {`{"pull":{"repository":"novox/mesh-catalog","number":434,"state":"open","head":"c948f3a61c74257e6e3147effb396d9d6434f491"}}`, true, false, "c948f3a61c74257e6e3147effb396d9d6434f491"}, + {`{"pull":{"state":"closed","head":"c948f3a61c74257e6e3147effb396d9d6434f491"}}`, false, false, "c948f3a61c74257e6e3147effb396d9d6434f491"}, } { - raw, err := os.ReadFile("testdata/captured/" + tc.file) - if err != nil { - t.Fatal(err) - } - content, _ := json.Marshal(map[string]any{"content": []map[string]any{{"type": "text", "text": string(raw)}}}) - ok, _ := json.Marshal(map[string]any{"ok": true, "output": "", "answer": json.RawMessage(raw)}) + raw := json.RawMessage(tc.answer) + content, _ := json.Marshal(map[string]any{"content": []map[string]any{{"type": "text", "text": tc.answer}}}) + ok, _ := json.Marshal(map[string]any{"ok": true, "output": "", "answer": raw}) for name, answer := range map[string]json.RawMessage{"bare": raw, "content list": content, "ok answer": ok} { - f := toolForge{ask: func(key string, body any) (json.RawMessage, error) { return answer, nil }} - open, merged, head, err := f.Pull("novox", "mesh-catalog", tc.number) + f := toolForge{ask: func(key string, body any) (json.RawMessage, error) { + asked.key, asked.body = key, body.(map[string]any) + return answer, nil + }} + open, merged, head, err := f.Pull("novox", "mesh-catalog", 434) if err != nil || open != tc.open || merged != tc.merged || head != tc.head { - t.Errorf("%s (%s): open %v merged %v head %q, err %v", tc.file, name, open, merged, head, err) + t.Errorf("%s (%s): open %v merged %v head %q, err %v", tc.answer, name, open, merged, head, err) + } + if asked.key != "seat:git.pull" || asked.body["repository"] != "novox/mesh-catalog" || asked.body["number"] != 434 { + t.Errorf("asked %+v", asked) } } } + failed := toolForge{ask: func(string, any) (json.RawMessage, error) { + return json.RawMessage(`{"error":"no such pull request","error-kind":"not-found"}`), nil + }} + if _, _, _, err := failed.Pull("novox", "mesh-catalog", 9); err == nil { + t.Error("an error answer was read as a pull request") + } + p, ok := pullEventOf([]byte(`{"repository":"novox/mesh-catalog","number":435,"branch":"fix/a","head":"7b5f7f84acb73aaf8189200548ff69fd9aac04b2","base":"main"}`)) + if !ok || p.Owner != "novox" || p.Repo != "mesh-catalog" || p.Number != 435 || p.HeadSHA != "7b5f7f84acb73aaf8189200548ff69fd9aac04b2" { + t.Errorf("%+v %v", p, ok) + } + if _, ok := pullEventOf([]byte(`{"repository":"norepo","number":1}`)); ok { + t.Error("a repository with no owner was read") + } } diff --git a/modules/mesh-config/cmd/mesh-config/pulls.go b/modules/mesh-config/cmd/mesh-config/pulls.go index 6fda9171a..50f01da94 100644 --- a/modules/mesh-config/cmd/mesh-config/pulls.go +++ b/modules/mesh-config/cmd/mesh-config/pulls.go @@ -480,14 +480,29 @@ func (c *Config) settlePull(pull string, merged bool, head string) []string { return said } -// PullEvent is the forge's `pull.merged` or `pull.closed`, as far as this module reads it. +// PullEvent is a pull request merged or closed, as far as this module reads it: the seat `git`'s `pull-merged` or `pull-closed` +// (design 47 G §654), read by pullEventOf. type PullEvent struct { - Owner string `json:"owner"` - Repo string `json:"repo"` - Number int `json:"number"` - Merged bool `json:"merged,omitempty"` - // HeadSHA is the head the pull request merged at (the forge's own word). - HeadSHA string `json:"head_sha,omitempty"` + Owner string + Repo string + Number int + Merged bool + // HeadSHA is the head the pull request merged at. + HeadSHA string +} + +// pullEventOf reads the seat `git`'s `pull-merged` or `pull-closed` body: the repository `/`, the number and the head. +func pullEventOf(body []byte) (PullEvent, bool) { + var e struct { + Repository string `json:"repository"` + Number int `json:"number"` + Head string `json:"head"` + } + if json.Unmarshal(body, &e) != nil { + return PullEvent{}, false + } + owner, repo, ok := strings.Cut(e.Repository, "/") + return PullEvent{Owner: owner, Repo: repo, Number: e.Number, HeadSHA: e.Head}, ok } func (p PullEvent) name() string { return fmt.Sprintf("%s/%s#%d", p.Owner, p.Repo, p.Number) } @@ -568,30 +583,32 @@ type toolForge struct { ask func(key string, body any) (json.RawMessage, error) } -// Pull asks the forge's gitea_get_pull_request whether a pull request is open, and whether it merged. +// Pull asks the seat `git`'s verb `pull` whether a pull request is open, and whether it merged. func (f toolForge) Pull(owner, repo string, number int) (bool, bool, string, error) { - raw, err := f.ask("gitea.gitea_get_pull_request", map[string]any{"owner": owner, "repo": repo, "number": number}) + raw, err := f.ask(GitPullAsk, map[string]any{"repository": owner + "/" + repo, "number": number}) if err != nil { return false, false, "", err } answer, output, ok := unwrapAnswer(raw) if !ok { - return false, false, "", fmt.Errorf("the forge refused gitea_get_pull_request: %s", output) + return false, false, "", fmt.Errorf("the seat git refused pull: %s", output) } var a struct { Pull struct { - State string `json:"state"` - Merged bool `json:"merged"` - // head is the branch name; the commit is head_sha (the forge module's mapPull). - HeadSHA string `json:"head_sha"` + State string `json:"state"` + Head string `json:"head"` } `json:"pull"` + Error string `json:"error"` } - if err := json.Unmarshal(answer, &a); err != nil || a.Pull.State == "" { - return false, false, "", fmt.Errorf("the forge's gitea_get_pull_request is not readable: %s", output) + if err := json.Unmarshal(answer, &a); err != nil || a.Error != "" || a.Pull.State == "" { + return false, false, "", fmt.Errorf("the seat git's pull is not readable: %s", output) } - return a.Pull.State == "open", a.Pull.Merged, a.Pull.HeadSHA, nil + return a.Pull.State == "open", a.Pull.State == "merged", a.Pull.Head, nil } +// GitPullAsk is the seat `git`'s verb that answers one pull request. +const GitPullAsk = "seat:git.pull" + // unwrapAnswer reads a tool's answer whatever wraps it: the protocol's content list, or {output, ok, answer}. func unwrapAnswer(raw json.RawMessage) (json.RawMessage, string, bool) { var s string diff --git a/modules/mesh-config/cmd/mesh-config/testdata/captured/README.md b/modules/mesh-config/cmd/mesh-config/testdata/captured/README.md deleted file mode 100644 index 6aa6b7020..000000000 --- a/modules/mesh-config/cmd/mesh-config/testdata/captured/README.md +++ /dev/null @@ -1,10 +0,0 @@ -Captured answers of the gitea module's `gitea_get_pull_request`, taken 2026-10-22 by fix-1826-catalog@shanks with -`mesh_call gitea.gitea_get_pull_request {"owner":"novox","repo":"mesh-catalog","number":N}` through the mesh MCP server. - -- get_pull_request_closed_merged_435.json: pull request 435, closed and merged. -- get_pull_request_open_434.json: pull request 434, open. - -Each file is the answer as the MCP server printed it. The runtime's own wrapping of the answer was not visible there -(the server prints the answer alone), so the tests also feed the files through the wrappings unwrapAnswer reads. -The `body` text is shortened to its first sentence and its Reviewed-by line (long prose of no interest here); every -other field is as returned. Note: `head` is the branch name, the commit is `head_sha`. diff --git a/modules/mesh-config/cmd/mesh-config/testdata/captured/get_pull_request_closed_merged_435.json b/modules/mesh-config/cmd/mesh-config/testdata/captured/get_pull_request_closed_merged_435.json deleted file mode 100644 index baff81a7f..000000000 --- a/modules/mesh-config/cmd/mesh-config/testdata/captured/get_pull_request_closed_merged_435.json +++ /dev/null @@ -1,17 +0,0 @@ -{ - "pull": { - "base": "main", - "body": "Part of design 47's milestone R4c-9 (ADR 0487): the image build module. A container image becomes one more artifact kind the builder builds through a build module, like Go and Node: mesh-build-image builds rootless with buildah, a candidate's check publishes nothing, a trunk build publishes exactly one digest and its build key names the reference. build-agent keeps building and delivering every image until R4g; nothing the operator sees changes. One delivery group on branch `feat/r4c9-image-build` in mesh-catalog and mesh-delivery. After the delivery the lead assigns buildah and mesh-build-image to shanks and ace.\n\nReviewed-by: reviewer-r4c9@shanks at 7b5f7f84acb73aaf8189200548ff69fd9aac04b2", - "head": "feat/r4c9-image-build", - "head_sha": "7b5f7f84acb73aaf8189200548ff69fd9aac04b2", - "html_url": "http://127.0.0.1:20000/novox/mesh-catalog/pulls/435", - "merge_commit_sha": "5a6d082e8c55603e47f074deb96da80fc9240ab4", - "merged": true, - "merged_at": "2026-10-22T08:57:10Z", - "number": 435, - "state": "closed", - "title": "Add the image build module: buildah and mesh-build-image, rootless, publishing only from the trunk (issues 1702, 1703, 1704, 1707)", - "updated_at": "2026-10-22T08:57:10Z", - "user": "mesh-tools" - } -} diff --git a/modules/mesh-config/cmd/mesh-config/testdata/captured/get_pull_request_open_434.json b/modules/mesh-config/cmd/mesh-config/testdata/captured/get_pull_request_open_434.json deleted file mode 100644 index f9975771f..000000000 --- a/modules/mesh-config/cmd/mesh-config/testdata/captured/get_pull_request_open_434.json +++ /dev/null @@ -1,15 +0,0 @@ -{ - "pull": { - "base": "main", - "body": "Part of design 47's milestone R4d: n8n and the bus run pinned upstream images (ADR 0480).\n\nReviewed-by: reviewer-r4d@shanks at c948f3a61c74257e6e3147effb396d9d6434f491\n", - "head": "feat/r4d", - "head_sha": "c948f3a61c74257e6e3147effb396d9d6434f491", - "html_url": "http://127.0.0.1:20000/novox/mesh-catalog/pulls/434", - "merged": false, - "number": 434, - "state": "open", - "title": "n8n and the bus run their upstream images with the mesh's programs mounted (issues 1639, 1640, 1641, 1642, ADR 0480 §1-§3, §5-§8; ADR 0497 §5)", - "updated_at": "2026-10-22T08:57:45Z", - "user": "mesh-tools" - } -} diff --git a/modules/mesh-config/module.json b/modules/mesh-config/module.json index b57f98b56..827c0f7cb 100644 --- a/modules/mesh-config/module.json +++ b/modules/mesh-config/module.json @@ -19,9 +19,8 @@ "operator-channel" ], "consumes": [ - "gitea.pull.merged", - "gitea.pull.closed", - "gitea.forge.pull-request.closed", + "git.pull-merged", + "git.pull-closed", "mesh-delivery-view.delivered", "node-engine.manifest-reads-changed", "mesh-catalog.manifest-accepted", @@ -37,7 +36,7 @@ "invokes": [ "seat:mesh-hand-acts.hand-act", "seat:mesh-hand-acts.warranted", - "gitea.gitea_get_pull_request", + "seat:git.pull", "mesh-delivery-view.node_haves", "seat:node-engine.manifest-reads", "mesh-delivery-indexer.indexer_config_readers", diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go index 86fb32541..651af2836 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go @@ -98,6 +98,7 @@ func main() { func run(h *Holder, l *listening) { <-stdio.Served() // Serve first: the state is reached through it, and an event is heard from as soon as it serves switchToRecords(h, l) + statePins(h, l) go h.Work() for _, pattern := range followed { pattern := pattern @@ -130,6 +131,20 @@ func switchToRecords(h *Holder, l *listening) { } } +// statePins says every pin the records hold once at each start (design 47 G §579), retrying until it can. +func statePins(h *Holder, l *listening) { + for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { + n, err := h.Records.StatePins() + if err == nil { + logf("[mesh-delivery] pins stated: %d", n) + return + } + l.set("not yet: " + err.Error()) + logf("[mesh-delivery] the pins cannot be stated yet (%v); asking again in %s", err, wait) + time.Sleep(wait) + } +} + // handle takes one event, then asks for a look at every step the event makes possible (Work, React). An event that cannot be read is // said and acknowledged: read again it would fail again. func handle(h *Holder, e stdio.Envelope) error { @@ -267,6 +282,7 @@ func tools(h *Holder, l *listening) []stdio.Tool { }}, } list = append(list, recordTools(h, meshhandactsclient.Recorder{Ask: stdio.Ask})...) + list = append(list, sourceTools(h, meshhandactsclient.Recorder{Ask: stdio.Ask})...) // A verb that changes a delivery asks for a look at what it makes possible (react.go). for i := range list { run := list[i].Run diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/recordevents.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/recordevents.go index 58488069a..cb123df7c 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/recordevents.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/recordevents.go @@ -255,3 +255,25 @@ func (rs *Records) sayEnded(r *Record) error { return rs.say(meshdelivery.SubjectRecordEnded, recordEnded{Record: r.Instance, Milestone: r.Milestone, ForgeMade: r.ForgeMade, Away: r.Ended.Away, At: r.Ended.At}, "record-ended."+r.Instance) } + +// StatePins says `pinned` again for every revision each record it keeps pins, then once `pins-stated` (design 47 G §579): the +// indexer's consumer of the seat's events hears pins said before it existed or older than the stream keeps, and the set form +// absorbs the repeats. Each repeat has an id of its own, since the bus keeps one message under one id inside its duplicate window. +func (rs *Records) StatePins() (int, error) { + kept, err := rs.Kept() + if err != nil { + return 0, err + } + at := rs.Now().UTC() + stated := 0 + for _, r := range kept { + for _, p := range sortedPins(r) { + id := "pinned." + r.Instance + "." + p.Revision + ".stated." + strconv.FormatInt(at.UnixNano(), 10) + if err := rs.say(meshdelivery.SubjectPinned, pinsOf(r, []PinnedRevision{p}, ""), id); err != nil { + return stated, err + } + stated++ + } + } + return stated, rs.say(meshdelivery.SubjectPinsStated, meshdelivery.PinsStatedEvent{At: at}, "pins-stated."+strconv.FormatInt(at.UnixNano(), 10)) +} diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/sourcetools.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/sourcetools.go new file mode 100644 index 000000000..2cebafb6c --- /dev/null +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/sourcetools.go @@ -0,0 +1,83 @@ +package main + +// The gate's tools over its source revisions (design 47 G §83, §84, §93): gate_source reads every branch's source revision +// whole, and gate_source_again says source-moved once more for one branch, recorded first in the hand-act log of the seat +// mesh-hand-acts, so a part that missed it reads the branch's source again. + +import ( + "errors" + "fmt" + "strconv" + "strings" + + "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" + seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/keys" +) + +// SourceAll is the seat verb source's answer for every branch the gate keeps a source revision of, each as the verb answers +// it for that branch, or why the branches cannot be read. +func (h *Holder) SourceAll() map[string]any { + list := h.SourceRevision("") + if list.Error != "" { + return map[string]any{"error": list.Error} + } + out := make([]seat.SourceAnswer, 0, len(list.Branches)) + for _, b := range list.Branches { + out = append(out, h.SourceRevision(b)) + } + return map[string]any{"branches": out} +} + +// SourceAgain says source-moved again for one branch with the revision its key holds now, after the hand-act log took it: +// nothing is said when the log did not take it, or the branch has no source key. +func (h *Holder) SourceAgain(branch, why, by string, acts handActs) (map[string]any, error) { + if strings.TrimSpace(branch) == "" || strings.TrimSpace(why) == "" { + return nil, errors.New("say the branch and why its source-moved is said again") + } + if h.Source == nil || h.SaySeat == nil { + return nil, errors.New("this gate keeps no source or says no events") + } + held, rev, err := h.Source.Read(branch) + if err != nil { + return nil, err + } + if held == nil { + return nil, fmt.Errorf("the gate keeps no source revision of the branch %s", branch) + } + if acts == nil { + return nil, errors.New("the hand-act log cannot be reached here, so nothing was said") + } + if _, err := acts.Record(meshhandactsclient.Request{What: "said source-moved again for the branch " + branch, + Why: why, Cause: "gate_source_again", Verb: "source-again"}); err != nil { + return nil, fmt.Errorf("the hand act was not recorded through mesh-hand-acts, so nothing was said: %w", err) + } + at := h.Now().UTC() + escaped := keys.EscapeName(branch) + id := "source-moved." + escaped + "." + strconv.FormatUint(rev, 10) + ".again." + strconv.FormatInt(at.UnixNano(), 10) + if err := h.SaySeat(seat.SubjectSourceMoved(escaped), + seat.SourceMovedEvent{Branch: branch, Group: held.Group, Revision: int64(rev), At: &at}, id); err != nil { + return nil, err + } + return map[string]any{"said": "source-moved", "branch": branch, "revision": rev, "by": by}, nil +} + +// sourceTools are the source's tools; acts is the hand-act log. +func sourceTools(h *Holder, acts handActs) []stdio.Tool { + return []stdio.Tool{ + {Name: "gate_source", + Description: "Every branch's source revision the gate keeps, each whole: its repositories with their trunk branch and head, " + + "the group that moved it last and the key's revision. Read, never written.", + Run: func(map[string]any) (any, error) { return h.SourceAll(), nil }}, + {Name: "gate_source_again", + Description: "Say source-moved again for one branch, with the revision its source key holds now, for a part that missed it; " + + "recorded in the hand-act log of the seat mesh-hand-acts first.", + Input: map[string]any{"type": "object", "properties": map[string]any{ + "branch": str("the watched branch"), "why": str("why it is said again, in plain words")}, + "required": []string{"branch", "why"}}, + Run: func(a map[string]any) (any, error) { + return h.SourceAgain(strArg(a, "branch"), strArg(a, "why"), stdio.CallerOf(a), acts) + }}, + } +} diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/sourcetools_test.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/sourcetools_test.go new file mode 100644 index 000000000..b5c142d9c --- /dev/null +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/sourcetools_test.go @@ -0,0 +1,86 @@ +package main + +import ( + "errors" + seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" + "strings" + "testing" + "time" +) + +type branchSource struct{ memSource } + +func (b *branchSource) Branches() ([]string, error) { + var out []string + for k := range b.vals { + out = append(out, k) + } + return out, nil +} + +// Design 47 G §83, §84: gate_source answers every branch's source revision whole; gate_source_again says source-moved once +// more for one branch with a message id of its own, only after the hand-act log took it. +func TestTheSourceToolsReadEveryBranchAndSayItAgainAfterTheLog(t *testing.T) { + src := &branchSource{} + for _, b := range []string{"main", "feat"} { + _, _ = src.Set(b, func(*Source) (Source, bool) { + return Source{Group: "g-" + b, Heads: map[string]string{"novox/app": "c1"}}, true + }) + } + type said struct{ subject, id string } + var events []said + now := time.Date(2026, 11, 12, 4, 0, 0, 0, time.UTC) + h := &Holder{Now: func() time.Time { return now }, Source: src, + SaySeat: func(subject string, body any, id string) error { + events = append(events, said{subject, id}) + return nil + }} + all := h.SourceAll()["branches"].([]seat.SourceAnswer) + if len(all) != 2 { + t.Fatalf("gate_source answered %+v", all) + } + acts := &handActsFake{refuse: errors.New("log down")} + if _, err := h.SourceAgain("main", "a part missed it", "me", acts); err == nil || len(events) != 0 { + t.Fatalf("said without the log: %v %v", err, events) + } + acts.refuse = nil + if _, err := h.SourceAgain("main", "", "me", acts); err == nil { + t.Fatal("said with no reason") + } + if _, err := h.SourceAgain("nothing", "x", "me", acts); err == nil { + t.Fatal("said for a branch with no source") + } + if _, err := h.SourceAgain("main", "a part missed it", "me", acts); err != nil { + t.Fatal(err) + } + if len(acts.took) != 1 || acts.took[0].Cause != "gate_source_again" || len(events) != 1 || + !strings.HasSuffix(events[0].subject, ".event.source-moved.main") || !strings.Contains(events[0].id, ".again.") { + t.Fatalf("log %+v events %+v", acts.took, events) + } +} + +// Design 47 G §579: at its start the gate says `pinned` again for every revision each record pins, each under an id of its own, +// then `pins-stated` once. +func TestTheGateStatesItsPinsThenSaysPinsStated(t *testing.T) { + at := time.Date(2026, 11, 12, 4, 0, 0, 0, time.UTC) + type said struct{ subject, id string } + var events []said + rs := &Records{Ledger: newLedgerFake(t), Now: func() time.Time { return at }, + Say: func(subject string, body any, id string) error { + events = append(events, said{subject, id}) + return nil + }} + r, _, err := rs.Open("47/G", map[string]string{"novox/app@1": "c1"}, false) + if err != nil { + t.Fatal(err) + } + if err := rs.Pin(r, HalfCatalog, 3); err != nil { + t.Fatal(err) + } + events = nil + n, err := rs.StatePins() + if err != nil || n != 1 || len(events) != 2 || !strings.HasSuffix(events[0].subject, ".event.pinned") || + !strings.Contains(events[0].id, ".stated.") || !strings.HasSuffix(events[1].subject, ".event.pins-stated") { + t.Fatalf("%d %v %+v", n, err, events) + } +} diff --git a/modules/mesh-delivery-gate/module.json b/modules/mesh-delivery-gate/module.json index e7d471e6e..0760ff7c9 100644 --- a/modules/mesh-delivery-gate/module.json +++ b/modules/mesh-delivery-gate/module.json @@ -349,7 +349,9 @@ "gate_record", "gate_record_end", "gate_record_repair", - "gate_records" + "gate_records", + "gate_source", + "gate_source_again" ], "config": { "record-retention": { diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/overview.go b/modules/mesh-knowledge/cmd/mesh-knowledge/overview.go index 9878134bd..88bbf9a39 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/overview.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/overview.go @@ -35,7 +35,7 @@ const ( // OverviewEvents are the events the page consumes, by source, as the manifest declares them. var OverviewEvents = map[string][]string{ SourceTracker: {"mesh-issues.opened", "mesh-issues.moved", "mesh-issues.milestone-set", "mesh-issues.milestone-moved"}, - SourceDelivery: {"mesh-delivery-gate.group", "mesh-delivery-gate.transition", "mesh-delivery-gate.merged"}, + SourceDelivery: {"git.pull-merged", "mesh-delivery-gate.transition", "mesh-pipeline-runner.group-ended"}, SourceAsks: {"messenger.offered", "messenger.closed"}, } @@ -344,6 +344,7 @@ func (o *Overview) Heard(source, event string, body json.RawMessage) { Design int `json:"design"` ID string `json:"id"` Group string `json:"group"` + Branch string `json:"branch"` Outcome string `json:"outcome"` Issues []int `json:"issues"` Members []string `json:"members"` @@ -383,18 +384,17 @@ func (o *Overview) Heard(source, event string, body json.RawMessage) { case SourceDelivery: id := firstNonEmpty(f.Group, f.ID) switch name { - case "group": - g := o.groups[id] - g.ID, g.State, g.Why = id, firstNonEmpty(f.State, f.To, g.State), firstNonEmpty(f.Why, g.Why) - g.Members = firstStrings(f.Members, g.Members) - o.groups[id] = g + case "pull-merged": + // A merged pull request leaves the group named by its head branch (a group's id is its branch name): it waits for + // the operator no longer (design 47 G §431). + delete(o.groups, f.Branch) case "transition": if g, ok := o.groups[id]; ok { g.State = firstNonEmpty(f.To, f.State, g.State) g.Why = firstNonEmpty(f.Why, g.Why) o.groups[id] = g } - case "merged": + case "group-ended": delete(o.groups, id) } case SourceAsks: diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/overview_test.go b/modules/mesh-knowledge/cmd/mesh-knowledge/overview_test.go index 262d4a0e6..646ac6f92 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/overview_test.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/overview_test.go @@ -25,7 +25,7 @@ func newOverviewWorld(t *testing.T) *overviewWorld { w := &overviewWorld{now: time.Date(2026, 11, 4, 12, 0, 0, 0, time.UTC), fail: map[string]bool{}, answers: map[string]string{ TrackerMilestonesVerb: `{"milestones":[{"design":47,"id":"1","outcome":"A merge is never dropped","status":"open","issues":[973,525]},{"design":47,"id":"2","outcome":"Done thing","status":"done","issues":[1]}]}`, TrackerListVerb: `[{"number":973,"title":"The window","status":"fixing"},{"number":525,"title":"The queue","status":"open"}]`, - DeliveryGroupsVerb: `[{"id":"fix/a","state":"held","why":"a member is held","members":["x@1"]}]`, + DeliveryGroupsVerb: `[{"id":"fix/a","state":"held","why":"a member is held","members":["x@1"]},{"id":"fix/b","state":"checking","members":["y@2"]}]`, AsksVerb: `{"asks":[{"key":"mesh-knowledge.1","asker":"mesh-knowledge","headline":"Approve ADR 0301","state":"open","offered_on":["telegram"]},{"key":"ask-0","asker":"x","headline":"old","state":"answered"}]}`, }} w.o = NewOverview(func(key string, body any) (json.RawMessage, error) { @@ -68,7 +68,7 @@ func TestThePageModelHasThreeBlocksKeptByEventsAndAsksNoVerbAgain(t *testing.T) t.Fatalf("the design block: %+v", m.Designs) } if len(m.Accepted) != 2 || len(m.Waiting) != 2 || len(m.Operator.Records) != 1 || m.Operator.Records[0].Record != "ADR 0303" || - len(m.Operator.Asks) != 1 || m.Operator.Asks[0].Key != "mesh-knowledge.1" || len(m.Operator.Groups) != 1 { + len(m.Operator.Asks) != 1 || m.Operator.Asks[0].Key != "mesh-knowledge.1" || len(m.Operator.Groups) != 2 { t.Fatalf("%+v", m) } asked := map[string]int{} @@ -76,20 +76,20 @@ func TestThePageModelHasThreeBlocksKeptByEventsAndAsksNoVerbAgain(t *testing.T) asked[v] = w.o.Calls(v) } // Events change the blocks. - w.o.Heard(SourceDelivery, "mesh-delivery-gate.group", json.RawMessage(`{"group":"fix/b","state":"checking","members":["y@2"]}`)) + w.o.Heard(SourceDelivery, "git.pull-merged", json.RawMessage(`{"repository":"novox/y","number":2,"branch":"fix/b"}`)) w.o.Heard(SourceDelivery, "mesh-delivery-gate.transition", json.RawMessage(`{"group":"fix/a","to":"approved"}`)) w.o.Heard(SourceAsks, "messenger.offered", json.RawMessage(`{"id":"2","asker":"mesh-delivery","headline":"Approve the group","channels":["telegram"]}`)) w.o.Heard(SourceTracker, "mesh-issues.opened", json.RawMessage(`{"number":1999,"title":"New","status":"open"}`)) w.o.Heard(SourceTracker, "mesh-issues.milestone-moved", json.RawMessage(`{"design":47,"id":"1","to":"done","issues":[973,525,1999]}`)) m, _ = w.o.Model() - if len(m.Operator.Groups) != 2 || m.Operator.Groups[0].State != "approved" || len(m.Operator.Asks) != 2 || len(m.Designs) != 0 { + if len(m.Operator.Groups) != 1 || m.Operator.Groups[0].State != "approved" || len(m.Operator.Asks) != 2 || len(m.Designs) != 0 { t.Fatalf("after the events: %+v", m) } - w.o.Heard(SourceDelivery, "mesh-delivery-gate.merged", json.RawMessage(`{"group":"fix/a"}`)) + w.o.Heard(SourceDelivery, "mesh-pipeline-runner.group-ended", json.RawMessage(`{"group":"fix/a","branch":"main","outcome":"passed"}`)) w.o.Heard(SourceAsks, "messenger.closed", json.RawMessage(`{"id":"1","asker":"mesh-knowledge","how":"answered"}`)) m, _ = w.o.Model() - if len(m.Operator.Groups) != 1 || m.Operator.Groups[0].ID != "fix/b" || len(m.Operator.Asks) != 1 || m.Operator.Asks[0].Key != "mesh-delivery.2" { - t.Fatalf("after merged and closed: %+v", m.Operator) + if len(m.Operator.Groups) != 0 || len(m.Operator.Asks) != 1 || m.Operator.Asks[0].Key != "mesh-delivery.2" { + t.Fatalf("after the group ended and the ask closed: %+v", m.Operator) } for v, n := range asked { if w.o.Calls(v) != n { @@ -146,7 +146,7 @@ func TestASourceThatDoesNotAnswerIsSaidAndAskedAgainOnABackoff(t *testing.T) { w.now = w.o.state[SourceDelivery].Mark.Next w.o.Due() m, _ = w.o.Model() - if len(w.o.Unanswered()) != 0 || len(m.Operator.Groups) != 1 { + if len(w.o.Unanswered()) != 0 || len(m.Operator.Groups) != 2 { t.Fatalf("%+v", m) } // A subscription that goes silent marks the source from the subscription and asks nothing while it is silent. diff --git a/modules/mesh-knowledge/module.json b/modules/mesh-knowledge/module.json index c765e2862..0382b6308 100644 --- a/modules/mesh-knowledge/module.json +++ b/modules/mesh-knowledge/module.json @@ -217,9 +217,9 @@ "mesh-issues.moved", "mesh-issues.milestone-set", "mesh-issues.milestone-moved", - "mesh-delivery-gate.group", + "git.pull-merged", "mesh-delivery-gate.transition", - "mesh-delivery-gate.merged", + "mesh-pipeline-runner.group-ended", "messenger.offered", "messenger.closed", "mesh-conditions.findings-asked" diff --git a/modules/mesh-recovery/cmd/mesh-recovery/dump.go b/modules/mesh-recovery/cmd/mesh-recovery/dump.go index 3fb2d9b83..3bb8d4a92 100644 --- a/modules/mesh-recovery/cmd/mesh-recovery/dump.go +++ b/modules/mesh-recovery/cmd/mesh-recovery/dump.go @@ -18,11 +18,11 @@ import ( // The seat node-backup's item events (design 47 G §554, §563), each kept last per subject on a stream of its own. const ( - backupSeat = "node-backup" - itemStarted = "item-started" - itemEnded = "item-ended" - itemStartedSteam = "SEAT_node-backup_item-started" - itemEndedStream = "SEAT_node-backup_item-ended" + backupSeat = "node-backup" + itemStarted = "item-started" + itemEnded = "item-ended" + itemStartedStream = "SEAT_node-backup_item-started" + itemEndedStream = "SEAT_node-backup_item-ended" ) func itemSubject(event, node, module, item string) string { @@ -103,7 +103,7 @@ func (d *dumpView) runs(module string) string { // readNow reads the last start and end of each of the bus module's items once more, through an ordered consumer made for // that read (§556): the restore said its restore-started, acknowledged, before it reads (§563). func (d *dumpView) readNow(ctx context.Context, module string) error { - for _, c := range []struct{ stream, event string }{{itemStartedSteam, itemStarted}, {itemEndedStream, itemEnded}} { + for _, c := range []struct{ stream, event string }{{itemStartedStream, itemStarted}, {itemEndedStream, itemEnded}} { cons, err := d.js.OrderedConsumer(ctx, c.stream, jetstream.OrderedConsumerConfig{ FilterSubjects: []string{itemSubject(c.event, "*", keys.EscapeName(module), "*")}, DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy}) diff --git a/modules/mesh-recovery/cmd/mesh-recovery/main.go b/modules/mesh-recovery/cmd/mesh-recovery/main.go index 3c50f938c..9e9a1a69e 100644 --- a/modules/mesh-recovery/cmd/mesh-recovery/main.go +++ b/modules/mesh-recovery/cmd/mesh-recovery/main.go @@ -128,7 +128,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err } dump := newDumpView(js, func() (string, error) { return seatlist.HolderOf(conn.Call, meshbroker.Seat) }) - for _, s := range []string{itemStartedSteam, itemEndedStream} { + for _, s := range []string{itemStartedStream, itemEndedStream} { if _, err := account.HearSeatEvents(ctx, js, s, dump.hear, logger.Printf); err != nil { return err } diff --git a/modules/mesh-recovery/cmd/mesh-recovery/whole_integration_test.go b/modules/mesh-recovery/cmd/mesh-recovery/whole_integration_test.go index 5cbaf0e31..1ca0f260b 100644 --- a/modules/mesh-recovery/cmd/mesh-recovery/whole_integration_test.go +++ b/modules/mesh-recovery/cmd/mesh-recovery/whole_integration_test.go @@ -91,7 +91,7 @@ func TestRecoveryWhole(t *testing.T) { } for stream, subject := range map[string]string{ nodeengine.StreamRunning: "mesh.seat.node-engine.event.running.>", - itemStartedSteam: "mesh.seat.node-backup.event.item-started.>", + itemStartedStream: "mesh.seat.node-backup.event.item-started.>", itemEndedStream: "mesh.seat.node-backup.event.item-ended.>", meshbroker.StreamSnapshotRestored: "mesh.seat.mesh-broker.event.snapshot-restored.>", } { diff --git a/modules/messenger/cmd/messenger/answers_test.go b/modules/messenger/cmd/messenger/answers_test.go index be28d5c0e..025eba1c4 100644 --- a/modules/messenger/cmd/messenger/answers_test.go +++ b/modules/messenger/cmd/messenger/answers_test.go @@ -5,6 +5,8 @@ import ( "fmt" "os" "path/filepath" + "reflect" + "sort" "strings" "sync" "testing" @@ -71,6 +73,13 @@ func (m *memAsks) UpdateAsk(k string, r AskRecord, rev uint64) (uint64, error) { m.asks[k], m.revs[k] = clone(r), m.rev return m.rev, nil } +func (m *memAsks) DeleteAsk(k string) error { + m.mu.Lock() + defer m.mu.Unlock() + delete(m.asks, k) + delete(m.revs, k) + return nil +} func (m *memAsks) AskKeys() ([]string, error) { m.mu.Lock() defer m.mu.Unlock() @@ -601,8 +610,17 @@ func TestAnAskIsRefusedInWordsToItsAsker(t *testing.T) { if strings.Contains(all, "d-6") || store.asks["mesh-delivery.d-6"].State != StateOpen || store.asks["mesh-delivery.d-3"].State != StateHeldBack { t.Errorf("the fourth ask was refused or the oldest not held back: d-6 %s, d-3 %s", store.asks["mesh-delivery.d-6"].State, store.asks["mesh-delivery.d-3"].State) } - if _, kept := store.asks["mesh-delivery.d-1"]; kept { - t.Error("a refused ask was recorded") + // A refused ask is kept as a closed record with its outcome and words (design 47 G §570), and `asks` lists it refused. + if rec, kept := store.asks["mesh-delivery.d-1"]; !kept || rec.State != string(occ.OutcomeRefused) || rec.Warrant == nil || + !strings.Contains(rec.Warrant.Words, "an ask that destroys lasts at most") || liveState(rec.State) { + t.Errorf("a refused ask was not kept as a closed record: %+v", rec) + } + listed := false + for _, v := range h.AsksNow() { + listed = listed || (v.Key == "mesh-delivery.d-1" && v.State == "refused" && strings.HasPrefix(v.Answer, "refused: ")) + } + if !listed { + t.Error("asks does not list the refused ask") } // A node's engine is an asker under node-engine- (novox/hq ADR 0563 §3, changing ADR 0259 §2): its ask is // recorded under that name, with its own bound of open asks; a name that is no asker's is dropped. @@ -1073,3 +1091,38 @@ func TestAnAsksMessageWithOneAnswerIsNotFoldedIntoADigest(t *testing.T) { t.Errorf("the ask's message was folded with another: %q", asked.Body) } } + +// Design 47 G §570: a refusal is written by a create only, so an open ask of the same id keeps its record; the twenty-first +// refusal of one asker removes its oldest, as the setting refused-kept says, each removal written in the journal. +func TestRefusedAsksAreKeptByACreateOnlyAndTheOldestGoesPastTheSetting(t *testing.T) { + h, _, _, c, store, _ := newRouter(t) + h.RefusedKept = func() int { return 3 } + open := heldAsk(c, "open-1") + submit(t, h, "mesh-delivery", open) + before := store.asks["mesh-delivery.open-1"] + same := heldAsk(c, "open-1") + same.Options[0].Level = occ.Destroy + submit(t, h, "mesh-delivery", same) + if got := store.asks["mesh-delivery.open-1"]; got.State != before.State || got.Warrant != nil { + t.Fatalf("a refusal of an open ask's id overwrote it: %+v", got) + } + for i := 1; i <= 5; i++ { + bad := heldAsk(c, fmt.Sprintf("bad-%d", i)) + bad.Options[0].Level = occ.Destroy + submit(t, h, "mesh-delivery", bad) + c.pass(time.Minute) + } + var kept []string + for k, r := range store.asks { + if r.State == string(occ.OutcomeRefused) { + kept = append(kept, k) + } + } + sort.Strings(kept) + if !reflect.DeepEqual(kept, []string{"mesh-delivery.bad-3", "mesh-delivery.bad-4", "mesh-delivery.bad-5"}) { + t.Fatalf("kept %v: the three newest refused asks of the asker", kept) + } + if h.askOf("mesh-delivery.bad-1") != nil { + t.Error("the router's own view still holds the removed refusal") + } +} diff --git a/modules/messenger/cmd/messenger/asks.go b/modules/messenger/cmd/messenger/asks.go index 0cba1d7b7..74a94cbf1 100644 --- a/modules/messenger/cmd/messenger/asks.go +++ b/modules/messenger/cmd/messenger/asks.go @@ -95,6 +95,8 @@ type Identity struct { type AskStore interface { GetAsk(key string) (*AskRecord, uint64, error) CreateAsk(key string, r AskRecord) (uint64, error) + // DeleteAsk removes one ask's record: the oldest refused one past the setting refused-kept (design 47 G §570). + DeleteAsk(key string) error UpdateAsk(key string, r AskRecord, revision uint64) (uint64, error) AskKeys() ([]string, error) GetIdentity(key string) (*Identity, error) @@ -864,10 +866,70 @@ func (h *Holder) offerAsk(r *Record) { h.flush(now, false) } -// refuseAsk says to its asker that an ask was not taken, and why. +// DefaultRefusedKept is how many refused asks of one asker the router keeps when the setting refused-kept gives none. +const DefaultRefusedKept = 20 + +// refuseAsk says to its asker that an ask was not taken, and why, and keeps it under `.` as a closed record with the +// outcome refused (design 47 G §570), by a create only: an id that holds a record already, an open ask with its tickets, is +// left as it is and the refusal is said on `decided` alone. func (h *Holder) refuseAsk(asker string, a oc.Ask, why string) { h.Logf("[messenger] refused an ask from %s (%s): %s", asker, a.ID, why) - h.tell(oc.Warrant{Ask: a.ID, Asker: asker, About: a.About, Outcome: occ.OutcomeRefused, Words: why, At: occ.FormatTime(h.Now())}) + now := h.Now() + w := oc.Warrant{Ask: a.ID, Asker: asker, About: a.About, Outcome: occ.OutcomeRefused, Words: why, At: occ.FormatTime(now)} + told := h.tell(w) == nil + if h.Asks == nil || !occ.UsableID(a.ID) { + return + } + key := askKey(asker, a.ID) + if h.askOf(key) != nil { + return + } + // Only its id and subject are kept: a refused ask's words may carry what may not leave the mesh (a secret's shape), so + // none of them is written to the record; its warrant says why in the router's words. + rec := AskRecord{Asker: asker, Ask: oc.Ask{ID: a.ID, About: a.About}, State: string(occ.OutcomeRefused), Opened: now, Closed: now, Warrant: &w, Told: told} + rev, err := h.Asks.CreateAsk(key, rec) + switch { + case errors.Is(err, ErrChanged): + return + case err != nil: + h.Logf("[messenger] the refusal of %s could not be kept: %v", key, err) + return + } + h.keepAsk(key, rec, rev) + h.pruneRefused(asker) +} + +// pruneRefused removes the oldest refused records of an asker past the setting refused-kept, each removal written in the +// journal (design 47 G §570). +func (h *Holder) pruneRefused(asker string) { + most := DefaultRefusedKept + if h.RefusedKept != nil { + most = h.RefusedKept() + } + h.mu.Lock() + var keys []string + for k, e := range h.asks { + if e.rec.Asker == asker && e.rec.State == string(occ.OutcomeRefused) { + keys = append(keys, k) + } + } + sort.Slice(keys, func(i, j int) bool { + a, b := h.asks[keys[i]].rec.Closed, h.asks[keys[j]].rec.Closed + return a.Before(b) || (a.Equal(b) && keys[i] < keys[j]) + }) + h.mu.Unlock() + for len(keys) > most { + oldest := keys[0] + keys = keys[1:] + if err := h.Asks.DeleteAsk(oldest); err != nil { + h.Logf("[messenger] the refused ask %s could not be removed: %v", oldest, err) + continue + } + h.mu.Lock() + delete(h.asks, oldest) + h.mu.Unlock() + h.Logf("[messenger] removed the oldest refused ask of %s, %s: %d are kept at most (refused-kept)", asker, oldest, most) + } } // tell publishes a warrant, or an ask's end, to its asker alone, deduplicated by the ask. diff --git a/modules/messenger/cmd/messenger/holder.go b/modules/messenger/cmd/messenger/holder.go index 30fe6dbdc..d87b60e68 100644 --- a/modules/messenger/cmd/messenger/holder.go +++ b/modules/messenger/cmd/messenger/holder.go @@ -330,6 +330,9 @@ type Holder struct { // CapPerHour is the most messages one channel gets an hour (the setting messages-per-hour, issue 1476); // nil, or a value below zero, is DefaultCapPerHour; 0 is no cap at all. TellCapPerHour is its own and stays fixed. CapPerHour func() int + // RefusedKept is how many refused asks of one asker are kept (the setting refused-kept, design 47 G §570); nil is + // DefaultRefusedKept. + RefusedKept func() int // Settle is how long a delivered walk holds what it touched after its end (the setting settle-seconds, novox/hq // ADR 0505 §3); nil is DefaultSettle. Settle func() time.Duration diff --git a/modules/messenger/cmd/messenger/main.go b/modules/messenger/cmd/messenger/main.go index e115da0f1..e74264b9f 100644 --- a/modules/messenger/cmd/messenger/main.go +++ b/modules/messenger/cmd/messenger/main.go @@ -39,6 +39,8 @@ type Settings struct { // SettleSeconds is how long a delivered walk still holds what it touched after its end (novox/hq ADR 0505 §3): // nil, not given, is DefaultSettle; 0 holds nothing after the end. SettleSeconds *int `json:"settle-seconds"` + // RefusedKept is how many refused asks of one asker the router keeps (design 47 G §570): nil, not given, is DefaultRefusedKept. + RefusedKept *int `json:"refused-kept"` } // zoneOf is the location a time-zone setting names, or this machine's own when it names none or one @@ -105,6 +107,17 @@ func readSettings(path string) (Settings, error) { default: return s, fmt.Errorf("settle-seconds is not a number") } + switch v := loose["refused-kept"].(type) { + case nil: + case float64: + n := int(v) + if v < 1 || v != float64(n) { + return s, fmt.Errorf("refused-kept is not a whole number of one or more") + } + s.RefusedKept = &n + default: + return s, fmt.Errorf("refused-kept is not a number") + } return s, nil } @@ -242,6 +255,7 @@ func (askStore) UpdateAsk(key string, r AskRecord, revision uint64) (uint64, err rev, err := stdio.State("asks").Update(key, r, revision) return rev, changed(err) } +func (askStore) DeleteAsk(key string) error { return stdio.State("asks").Delete(key) } func (askStore) AskKeys() ([]string, error) { return stdio.State("asks").Keys() } func (askStore) GetIdentity(key string) (*Identity, error) { e, err := stdio.State("identities").Get(key) @@ -366,6 +380,12 @@ func main() { } return DefaultCapPerHour }, + RefusedKept: func() int { + if n := settings().RefusedKept; n != nil { + return *n + } + return DefaultRefusedKept + }, Settle: func() time.Duration { if n := settings().SettleSeconds; n != nil && *n >= 0 { return time.Duration(*n) * time.Second diff --git a/modules/messenger/cmd/messenger/whole_test.go b/modules/messenger/cmd/messenger/whole_test.go index 68b3801d3..e8cc5a6a0 100644 --- a/modules/messenger/cmd/messenger/whole_test.go +++ b/modules/messenger/cmd/messenger/whole_test.go @@ -149,8 +149,9 @@ func TestAnAskWhoseWholeWordsCarryASecretIsRefused(t *testing.T) { if len(tg.sends) != 0 || len(got) != 1 || got[0].Outcome != occ.OutcomeRefused || !strings.Contains(got[0].Words, "whole words carry secret") { t.Errorf("%q: sent %d, told %+v", secret, len(tg.sends), got) } - if _, kept := store.asks["mesh-controller.s-4"]; kept { - t.Errorf("%q: the refused ask was recorded", secret) + if rec, kept := store.asks["mesh-controller.s-4"]; !kept || rec.State != "refused" || rec.Ask.Whole != "" || rec.Ask.Headline != "" || + strings.Contains(fmt.Sprint(rec), "ghp_") || strings.Contains(fmt.Sprint(rec), "hunter2") { + t.Errorf("%q: the refused ask was not kept as a closed record without its words: %+v", secret, rec) } } } @@ -204,7 +205,7 @@ func TestEveryKeyAndItsWholeValueIsInTheNoteWhereTheSenderIsProven(t *testing.T) b.WholeDetails = "+ token: ghp_abcdefghijklmnopqrstuvwxyz0123456789" submit(t, h, "mesh-controller", b) settle(h, c) - if _, kept := store.asks["mesh-controller.s-10"]; kept || len(tg.sends) != 1 { + if rec, kept := store.asks["mesh-controller.s-10"]; (kept && rec.State != "refused") || rec.Ask.WholeDetails != "" || len(tg.sends) != 1 { t.Errorf("an ask whose whole details carry a secret was taken") } } diff --git a/modules/messenger/files/settings.json b/modules/messenger/files/settings.json index d628295dc..7eaa42a2d 100644 --- a/modules/messenger/files/settings.json +++ b/modules/messenger/files/settings.json @@ -3,5 +3,6 @@ "only-what-needs-me": [], "messages-per-hour": 100, "settle-seconds": 120, + "refused-kept": 20, "readers": {"offered": ["mesh-knowledge"], "closed": ["mesh-knowledge"]} } diff --git a/modules/messenger/module.json b/modules/messenger/module.json index c6b1e56de..f3b66dd47 100644 --- a/modules/messenger/module.json +++ b/modules/messenger/module.json @@ -164,6 +164,14 @@ "trusted": false, "default": 100 }, + "refused-kept": { + "type": "number", + "description": "how many refused asks of one asker the router keeps as closed records, a whole number from 1 to 1000: the twenty-first refusal of an asker removes its oldest (design 47 G section 570).", + "minimum": 1, + "maximum": 1000, + "trusted": false, + "default": 20 + }, "settle-seconds": { "type": "number", "description": "seconds a delivery that ended well still holds the messages about what it changed, so a problem that passes as the delivery settles is not sent (ADR 0505 §3); 0 holds nothing after the end.", @@ -194,7 +202,7 @@ "type": "file", "path": "${dir:state}/settings.json", "mode": "0600", - "takes-config": ["time-zone", "desktop-machines", "only-what-needs-me", "messages-per-hour", "settle-seconds", "readers"], + "takes-config": ["time-zone", "desktop-machines", "only-what-needs-me", "messages-per-hour", "settle-seconds", "refused-kept", "readers"], "merge": "json", "template": "files/settings.json", "owner": "messenger" diff --git a/modules/records/README.md b/modules/records/README.md index 6fb679cd7..3a90323ea 100644 --- a/modules/records/README.md +++ b/modules/records/README.md @@ -8,7 +8,7 @@ whoever holds the console sees `records_search` beside every other tool and a sy in the design record without knowing it is there. **A checkout, not a copy.** The same bytes the repository holds, at a commit every answer names, -brought up to date on every merge the forge announces (`gitea.pull.merged`) and every ten minutes +brought up to date on every merge the forge announces (`git.pull-merged`) and every ten minutes besides. Nothing is indexed, transformed or summarised, so nothing can drift from the source except by lagging behind it, and the lag is in `records_status`. diff --git a/modules/records/cmd/records/main.go b/modules/records/cmd/records/main.go index 718120d66..18b2ceda7 100644 --- a/modules/records/cmd/records/main.go +++ b/modules/records/cmd/records/main.go @@ -109,16 +109,11 @@ func keep(r *Records) { } }() - // A merge on the forge into the repository this reads: pull now. The event names the repository by - // owner and name (gitea's `pull.merged`); anything else is somebody else's merge. + // A merge into the repository this reads: pull now. The seat `git`'s `pull-merged` names the repository `/` + // (design 47 G §654); anything else is somebody else's merge. subscribe := func() error { - return stdio.Subscribe("gitea.pull.merged", func(e stdio.Envelope) error { - var body struct { - Owner string `json:"owner"` - Repo string `json:"repo"` - } - _ = json.Unmarshal(e.Body, &body) - if body.Owner+"/"+body.Repo != r.Repository { + return stdio.Subscribe("git.pull-merged", func(e stdio.Envelope) error { + if !mergedInto(e.Body, r.Repository) { return nil } fmt.Fprintf(os.Stderr, "[records] %s merged; syncing\n", r.Repository) @@ -136,6 +131,14 @@ func keep(r *Records) { } } +// mergedInto is whether a `pull-merged` body names this repository, `/`. +func mergedInto(body []byte, repository string) bool { + var e struct { + Repository string `json:"repository"` + } + return json.Unmarshal(body, &e) == nil && e.Repository != "" && e.Repository == repository +} + func main() { r, err := FromEnv(os.Getenv) if err != nil { diff --git a/modules/records/cmd/records/records_test.go b/modules/records/cmd/records/records_test.go index 2dab8135a..d4d13b4a2 100644 --- a/modules/records/cmd/records/records_test.go +++ b/modules/records/cmd/records/records_test.go @@ -187,3 +187,18 @@ func TestRecordsDecisionsNamesTheKnowledgeBaseAndReadsNothingOfHq(t *testing.T) t.Errorf("its description does not say where the decision records are: %s", decisions.Description) } } + +// Design 47 G §654: a merge is heard from the seat git's pull-merged, which names the repository `/`. +func TestAMergeIntoThisRepositoryIsRecognisedFromTheSeatGitEvent(t *testing.T) { + for body, want := range map[string]bool{ + `{"repository":"novox/hq","number":3}`: true, + `{"repository":"novox/mesh-catalog"}`: false, + `{"owner":"novox","repo":"hq"}`: false, + `not json`: false, + `{"repository":""}`: false, + } { + if got := mergedInto([]byte(body), "novox/hq"); got != want { + t.Errorf("%s: %v, want %v", body, got, want) + } + } +} diff --git a/modules/records/module.json b/modules/records/module.json index 4a040a3b0..3e23fe06d 100644 --- a/modules/records/module.json +++ b/modules/records/module.json @@ -9,7 +9,7 @@ "git": "${dir:mesh-state}/git.json" }, "consumes": [ - "gitea.pull.merged" + "git.pull-merged" ], "tools": [ "records_search", diff --git a/modules/restic/cmd/restic-backups/backups.go b/modules/restic/cmd/restic-backups/backups.go index ff33e5036..14cbfd915 100644 --- a/modules/restic/cmd/restic-backups/backups.go +++ b/modules/restic/cmd/restic-backups/backups.go @@ -208,6 +208,15 @@ type Backups struct { // measuring run as processes of their own). Node is the machine's name, said in the events. Emit func(event string, body any) error Node string + // Bus says the seat node-backup's item events and reads the recovery module's restores (items.go); nil where this process has + // no bus account. BrokerModule is the module that holds the seat mesh-broker now, read from the seat list. + Bus ItemBus + BrokerModule func() (string, error) + // PutOffs is the module's own state `put-off`, whose keys are the items put off for a restore; PutOffBound is the setting + // put-off-bound; Findings is the finding writer of the night's process (putoff.go). Each nil where the process has none. + PutOffs PutOffState + PutOffBound time.Duration + Findings Findings // Readings is every reading disk-load keeps for this machine (its state `readings`, ADR 0571 clause 3); nil where the // process has no runtime to read it. Readings func() ([]diskreadings.Reading, error) @@ -332,8 +341,21 @@ func (b *Backups) ensureRepository(ctx context.Context) error { // one is one module's night: its run lines, then one snapshot of its paths. A failure is that // module's alone. -func (b *Backups) one(ctx context.Context, d Declared, hand bool) Night { - n := Night{At: b.Now().UTC()} +func (b *Backups) one(ctx context.Context, d Declared, hand bool) (n Night) { + n = Night{At: b.Now().UTC()} + ended, err := b.beginItems(ctx, d.Module) + if err != nil { + n.Error = err.Error() + b.Say("%s: NOT backed up: %s", d.Module, n.Error) + return n + } + defer func() { + if n.OK { + ended(nil) + } else { + ended(errors.New(n.Error)) + } + }() fail := func(err error) Night { n.Error = err.Error() b.Say("%s: NOT backed up: %s", d.Module, n.Error) @@ -609,7 +631,7 @@ func (b *Backups) BackedUp(ctx context.Context, module string) ([]ModuleBackups, } listed[d.Module] = true m := ModuleBackups{Module: d.Module, Runs: len(d.Runs), Paths: d.Paths, Retention: inForce, Thinned: thinned, - Data: itemReports(items[d.Module], measured[d.Module], d.Paths, good[d.Module]), + Data: b.withPutOff(d.Module, itemReports(items[d.Module], measured[d.Module], d.Paths, good[d.Module])), HandTaken: handKept(d.Module, snaps, goodNights, inForce)} if n, ok := nights[d.Module]; ok { m.LastNight = &n @@ -628,7 +650,7 @@ func (b *Backups) BackedUp(ctx context.Context, module string) ([]ModuleBackups, if listed[name] || (module != "" && name != module) { continue } - out = append(out, ModuleBackups{Module: name, Paths: []string{}, HandTaken: []HandKept{}, Data: itemReports(its, measured[name], nil, time.Time{})}) + out = append(out, ModuleBackups{Module: name, Paths: []string{}, HandTaken: []HandKept{}, Data: b.withPutOff(name, itemReports(its, measured[name], nil, time.Time{}))}) } return out, nil } diff --git a/modules/restic/cmd/restic-backups/data.go b/modules/restic/cmd/restic-backups/data.go index 950146ebb..56e4ce7c9 100644 --- a/modules/restic/cmd/restic-backups/data.go +++ b/modules/restic/cmd/restic-backups/data.go @@ -37,6 +37,8 @@ type ItemReport struct { Peak *int64 `json:"peak_bytes,omitempty"` PeakDay string `json:"peak_day,omitempty"` LastBackup *time.Time `json:"last_backup,omitempty"` + // PutOff is set while the item waits for a restore of the recovery module: its act and since when (§563). + PutOff *PutOff `json:"put_off,omitempty"` } var safePath = regexp.MustCompile(`^/[^\s'"\\$` + "`" + `]*$`) diff --git a/modules/restic/cmd/restic-backups/itembus.go b/modules/restic/cmd/restic-backups/itembus.go new file mode 100644 index 000000000..cd0d95a41 --- /dev/null +++ b/modules/restic/cmd/restic-backups/itembus.go @@ -0,0 +1,315 @@ +package main + +// The live bus of the item events (items.go): the module's own broker account, as the recovery module's process holds its own +// (design 47 G A-150: the process `night` has a bus account of its own, says its items and reads the recovery module's restores). + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "sync" + "sync/atomic" + "time" + + "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/keys" + "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/seatlist" + "git.novox.be/novox/mesh-sdk/go/views" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +// The streams and subjects of the seats' events the holder reads and says. +const ( + backupSeat = "node-backup" + recoverySeat = "mesh-recovery" + streamItemStarted = "SEAT_node-backup_item-started" + streamItemEnded = "SEAT_node-backup_item-ended" + streamRestoreStarted = "SEAT_mesh-recovery_restore-started" + streamRestored = "SEAT_mesh-recovery_restored" + readWait = 5 * time.Second +) + +func recoveryEventSubject(event, module, kept string) string { + return "mesh.seat." + recoverySeat + ".event." + event + "." + module + "." + kept +} + +func itemEventSubject(event, node, module, item string) string { + return "mesh.seat." + backupSeat + ".event." + event + "." + node + "." + module + "." + item +} + +// liveBus is ItemBus over the module's own account. +type liveBus struct { + conn *process.Conn + nc *nats.Conn + js jetstream.JetStream + node string + account process.Account +} + +// connectItems opens the bus as the module's own account when the process was given its credential (MESH_BROKER_FILE); nil, nil +// when it was not. +func connectItems(getenv func(string) string) (*liveBus, error) { + if strings.TrimSpace(getenv(process.BrokerFileEnv)) == "" { + return nil, nil + } + account, err := process.ReadBrokerAccount(getenv) + if err != nil { + return nil, err + } + conn, err := process.Connect(account) + if err != nil { + return nil, err + } + nc, js, err := account.Dial() + if err != nil { + conn.Close() + return nil, err + } + return &liveBus{conn: conn, nc: nc, js: js, node: account.Node, account: account}, nil +} + +// Broker answers how to find the holder of the seat mesh-broker. +func (l *liveBus) Broker() (string, error) { return seatlist.HolderOf(l.conn.Call, BrokerSeat) } + +// Close ends both connections. +func (l *liveBus) Close() { l.nc.Close(); l.conn.Close() } + +// PutOffs is the module's own state `put-off`, its keys with their own time to live (A-155). +func (l *liveBus) PutOffs() PutOffState { return stateOfPutOffs{l.conn.State(PutOffStateName)} } + +// Findings starts the module's held finding writer over its state `findings` (§561), as the recovery module does (§461): started +// before the night's first backup, so what it writes is held and written. It answers how to close it. +func (l *liveBus) Findings(ctx context.Context, logf func(string, ...any)) (Findings, func(), error) { + w, err := views.NewHeldWriter(views.WriterConfig[meshconditions.Finding, meshconditions.FindingCleared, meshconditions.FindingsStated, meshconditions.FindingsAskedEvent]{ + Asked: meshconditions.SubjectFindingsAsked, Module: l.account.Module, Node: l.account.Node, State: l.conn.State("findings"), + Kinds: []string{FindingBackupWaits}, JS: l.js, Conn: l.nc, Logf: logf}) + if err != nil { + return nil, nil, err + } + if err := w.Start(ctx); err != nil { + return nil, nil, err + } + return w, w.Close, nil +} + +func (l *liveBus) Say(event string, ref ItemRef, body map[string]any) error { + return l.conn.EmitSeat(backupSeat, event+"."+l.node+"."+keys.EscapeName(ref.Module)+"."+keys.EscapeName(ref.Item), body) +} + +// last reads every subject's last message of a stream through an ordered consumer made for this one read (§556). +func (l *liveBus) last(ctx context.Context, stream, filter string, take func(subject string, data []byte)) error { + cons, err := l.js.OrderedConsumer(ctx, stream, jetstream.OrderedConsumerConfig{ + FilterSubjects: []string{filter}, DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy}) + if err != nil { + return fmt.Errorf("the read of %s: %w", stream, err) + } + info, err := cons.Info(ctx) + if err != nil { + return fmt.Errorf("the read of %s: %w", stream, err) + } + for left := info.NumPending; left > 0; { + m, err := cons.Next(jetstream.FetchMaxWait(readWait)) + if err != nil { + return fmt.Errorf("the read of %s: %w", stream, err) + } + take(m.Subject(), m.Data()) + meta, err := m.Metadata() + if err != nil { + return err + } + left = meta.NumPending + } + return nil +} + +func atOf(data []byte) time.Time { + var b struct { + At time.Time `json:"at"` + } + _ = json.Unmarshal(data, &b) + return b.At +} + +func (l *liveBus) Unended() ([]ItemRef, error) { + ctx, cancel := context.WithTimeout(context.Background(), 2*readWait) + defer cancel() + started, ended := map[ItemRef]time.Time{}, map[ItemRef]time.Time{} + for _, c := range []struct { + stream, event string + into map[ItemRef]time.Time + }{{streamItemStarted, EventItemStarted, started}, {streamItemEnded, EventItemEnded, ended}} { + prefix := itemEventSubject(c.event, l.node, "", "") + prefix = strings.TrimSuffix(prefix, "..") + c := c + err := l.last(ctx, c.stream, itemEventSubject(c.event, l.node, "*", "*"), func(subject string, data []byte) { + module, item, ok := strings.Cut(strings.TrimPrefix(subject, prefix+"."), ".") + if !ok { + return + } + m, e1 := keys.UnescapeName(module) + i, e2 := keys.UnescapeName(item) + if e1 == nil && e2 == nil { + c.into[ItemRef{m, i}] = atOf(data) + } + }) + if err != nil { + return nil, err + } + } + var out []ItemRef + for ref, at := range started { + if end, ok := ended[ref]; !ok || end.Before(at) { + out = append(out, ref) + } + } + sortRefs(out) + return out, nil +} + +func (l *liveBus) Restoring(module string) (string, error) { + ctx, cancel := context.WithTimeout(context.Background(), 2*readWait) + defer cancel() + type act struct { + id string + at time.Time + } + started, restored := map[string]act{}, map[string]act{} + for _, c := range []struct { + stream, event string + into map[string]act + }{{streamRestoreStarted, "restore-started", started}, {streamRestored, "restored", restored}} { + prefix := recoveryEventSubject(c.event, keys.EscapeName(module), "") + c := c + err := l.last(ctx, c.stream, recoveryEventSubject(c.event, keys.EscapeName(module), "*"), func(subject string, data []byte) { + var b struct { + Act string `json:"act"` + At time.Time `json:"at"` + } + if json.Unmarshal(data, &b) == nil { + c.into[strings.TrimPrefix(subject, prefix)] = act{b.Act, b.At} + } + }) + if err != nil { + return "", err + } + } + var running []string + for kept, s := range started { + if r, ok := restored[kept]; !ok || r.id != s.id { + running = append(running, s.id) + } + } + if len(running) == 0 { + return "", nil + } + if len(running) > 1 { + sortStrings(running) + } + return running[0], nil +} + +func (l *liveBus) WaitRestored(ctx context.Context, want string) error { + cons, err := l.js.OrderedConsumer(ctx, streamRestored, jetstream.OrderedConsumerConfig{ + FilterSubjects: []string{"mesh.seat." + recoverySeat + ".event.restored.*.*"}, DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy}) + if err != nil { + return err + } + done := make(chan struct{}) + var once sync.Once + cc, err := cons.Consume(func(m jetstream.Msg) { + var b struct { + Act string `json:"act"` + } + if json.Unmarshal(m.Data(), &b) == nil && b.Act == want { + once.Do(func() { close(done) }) + } + }) + if err != nil { + return err + } + defer cc.Stop() + // The wait ends on the restored or when ctx ends (the bound's expiry cancels it), never on a fetch's own time. + select { + case <-done: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +func sortStrings(s []string) { + for i := 1; i < len(s); i++ { + for j := i; j > 0 && s[j] < s[j-1]; j-- { + s[j], s[j-1] = s[j-1], s[j] + } + } +} + +// PutOffStateName is the module's own state of the items put off for a restore, whose keys have a time to live (key-ttl). +const PutOffStateName = "put-off" + +// putOffBucket is the state's calls this module makes on its bucket. +type putOffBucket interface { + CreateTTL(key string, value any, ttl time.Duration) (uint64, error) + Watch(key string, handle func(stdio.StateChange) error) error + Delete(key string) error + Keys() ([]string, error) + Get(key string) (*stdio.StateEntry, error) +} + +// errBoundEnded ends a bound key's watch once the key is gone. +var errBoundEnded = errors.New("the bound key is removed") + +// stateOfPutOffs times the wait for a restore by a bound key of the state put-off (A-77, A-155): the key is watched before it is +// written, so its expiry is never missed, and the watch ends with the key's removal. +type stateOfPutOffs struct{ s putOffBucket } + +func (p stateOfPutOffs) Hold(key string, rec PutOff, ttl time.Duration, expired func()) (func(), error) { + var ended atomic.Bool + if err := p.s.Watch(key, func(c stdio.StateChange) error { + if c.Op == "put" || c.Current { + return nil + } + if !ended.Swap(true) { + expired() + } + return errBoundEnded + }); err != nil { + return nil, err + } + if _, err := p.s.CreateTTL(key, rec, ttl); err != nil { + ended.Store(true) + _ = p.s.Delete(key) + return nil, err + } + return func() { + if !ended.Swap(true) { + _ = p.s.Delete(key) + } + }, nil +} + +func (p stateOfPutOffs) All() (map[string]PutOff, error) { + names, err := p.s.Keys() + if err != nil { + return nil, err + } + out := map[string]PutOff{} + for _, k := range names { + e, err := p.s.Get(k) + if err != nil || e == nil { + continue + } + var rec PutOff + if json.Unmarshal(e.Value, &rec) == nil { + out[k] = rec + } + } + return out, nil +} diff --git a/modules/restic/cmd/restic-backups/items.go b/modules/restic/cmd/restic-backups/items.go new file mode 100644 index 000000000..cbe87f2a1 --- /dev/null +++ b/modules/restic/cmd/restic-backups/items.go @@ -0,0 +1,212 @@ +package main + +// The backup holder's item events and its give-way to a restore of the recovery module (design 47 G §554, §556, §563): each item +// of a module is said started and ended on the seat node-backup; for the items of the holder of the seat mesh-broker the holder +// says its start, then reads the recovery module's restore-started and restored once more, and when a restore runs it ends the +// item put off, naming that restore's act, and starts it again on that act's restored. Said means acknowledged. + +import ( + "context" + "errors" + "fmt" + "sort" + "strings" + "sync" +) + +// The seat node-backup's events of an item, and the seat of the holder whose items are the bus's. +const ( + EventItemStarted = "item-started" + EventItemEnded = "item-ended" + BrokerSeat = "mesh-broker" +) + +// The item's outcomes on item-ended. +const ( + ItemDone = "done" + ItemFailed = "failed" + ItemPutOff = "put-off" +) + +// StoppedWhy is the error of an item said started and never ended, ended at the holder's next start. +const StoppedWhy = "stopped" + +// ItemRef is one item of a module on a node. +type ItemRef struct{ Module, Item string } + +// ItemBus is what the backup holder says and reads on the bus. +type ItemBus interface { + // Say publishes one item event, `item-started` or `item-ended`, and answers once the bus acknowledged it. + Say(event string, ref ItemRef, body map[string]any) error + // Unended are this node's items said started whose last start is newer than their last end, read last per subject (§556). + Unended() ([]ItemRef, error) + // Restoring reads the recovery module's restore-started and restored of a module's kept data once more, last per subject, and + // answers the act of a restore that runs, "" when none does. + Restoring(module string) (string, error) + // WaitRestored answers once the recovery module said restored with the act. + WaitRestored(ctx context.Context, act string) error +} + +// ItemsOf are the items of a module a backup covers: the data items it declares that are not a cache. +func (b *Backups) ItemsOf(module string) []ItemRef { + all, err := b.Items() + if err != nil { + b.Say("the data items of %s could not be read: %v", module, err) + return nil + } + var out []ItemRef + for _, it := range all[module] { + if it.Class == "cache" { + continue + } + out = append(out, ItemRef{Module: module, Item: it.ID}) + } + return out +} + +func (b *Backups) sayItem(event string, ref ItemRef, body map[string]any) error { + body["at"] = b.Now().UTC() + return b.Bus.Say(event, ref, body) +} + +// StopStale ends every item this node said started and never ended with `failed` and the error `stopped`, so no restore waits on +// a dump that no longer runs (§563). It holds the night's lock, so no backup of this module is running meanwhile. +func (b *Backups) StopStale() error { + if b.Bus == nil { + return nil + } + release, err := b.hold(nightLock) + if err != nil { + return err + } + defer release() + left, err := b.Bus.Unended() + if err != nil { + return fmt.Errorf("the items said started and never ended could not be read: %w", err) + } + var failed []string + for _, ref := range left { + if err := b.sayItem(EventItemEnded, ref, map[string]any{"outcome": ItemFailed, "error": StoppedWhy}); err != nil { + failed = append(failed, ref.Module+"."+ref.Item+": "+err.Error()) + } + } + if len(failed) > 0 { + return errors.New("items left started could not be ended: " + strings.Join(failed, "; ")) + } + return nil +} + +// beginItems says every item of a module started and, for the items of the holder of the seat mesh-broker, gives way to a restore +// that runs: its items are ended put off naming the act, wait for that restore's restored, and are started again. It answers how to +// end the items when the backup ends. A module with no item says nothing. +func (b *Backups) beginItems(ctx context.Context, module string) (func(err error), error) { + nothing := func(error) {} + if b.Bus == nil { + return nothing, nil + } + items := b.ItemsOf(module) + if len(items) == 0 { + return nothing, nil + } + started := func() error { + for _, ref := range items { + if err := b.sayItem(EventItemStarted, ref, map[string]any{}); err != nil { + return err + } + } + return nil + } + end := func(outcome string, extra map[string]any) error { + var first error + for _, ref := range items { + body := map[string]any{"outcome": outcome} + for k, v := range extra { + body[k] = v + } + if err := b.sayItem(EventItemEnded, ref, body); err != nil && first == nil { + first = err + } + } + return first + } + finish := func(err error) { + var said error + if err != nil { + said = end(ItemFailed, map[string]any{"error": err.Error()}) + } else { + said = end(ItemDone, nil) + } + if said != nil { + b.Say("the end of %s's items could not be said: %v", module, said) + } + } + if err := started(); err != nil { + return nothing, fmt.Errorf("the start of %s's items could not be said: %w", module, err) + } + b.clearWaits(items) + if b.BrokerModule == nil { + return finish, nil + } + broker, err := b.BrokerModule() + if err != nil { + _ = end(ItemFailed, map[string]any{"error": err.Error()}) + return nothing, fmt.Errorf("who holds the seat %s is not known: %w", BrokerSeat, err) + } + if broker != module { + return finish, nil + } + for { + act, err := b.Bus.Restoring(module) + if err != nil { + _ = end(ItemFailed, map[string]any{"error": err.Error()}) + return nothing, fmt.Errorf("the recovery module's restores could not be read: %w", err) + } + if act == "" { + return finish, nil + } + b.Say("%s's backup gives way to the restore %s of the recovery module", module, act) + if err := end(ItemPutOff, map[string]any{"restore": act}); err != nil { + return nothing, fmt.Errorf("the put-off of %s's items could not be said: %w", module, err) + } + wctx, stop := context.WithCancel(ctx) + var expiredMu sync.Mutex + expired := false + release, err := b.keepPutOff(items, act, func(ref ItemRef) { + expiredMu.Lock() + expired = true + expiredMu.Unlock() + b.writeWaits(ref, act) + stop() + }) + if err != nil { + stop() + return nothing, err + } + waitErr := b.Bus.WaitRestored(wctx, act) + release() + stop() + expiredMu.Lock() + ran := expired + expiredMu.Unlock() + if ran { + return nothing, fmt.Errorf("%w: the restore %s", errPutOffExpired, act) + } + if waitErr != nil { + return nothing, fmt.Errorf("the end of the restore %s could not be waited for: %w", act, waitErr) + } + if err := started(); err != nil { + return nothing, fmt.Errorf("the start of %s's items could not be said: %w", module, err) + } + b.clearWaits(items) + } +} + +// sortRefs orders items by module, then item. +func sortRefs(refs []ItemRef) { + sort.Slice(refs, func(i, j int) bool { + if refs[i].Module != refs[j].Module { + return refs[i].Module < refs[j].Module + } + return refs[i].Item < refs[j].Item + }) +} diff --git a/modules/restic/cmd/restic-backups/items_test.go b/modules/restic/cmd/restic-backups/items_test.go new file mode 100644 index 000000000..aabae9fb1 --- /dev/null +++ b/modules/restic/cmd/restic-backups/items_test.go @@ -0,0 +1,258 @@ +package main + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "reflect" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" +) + +// itemBusFake records every call, in order, as the backup holder makes them. +type itemBusFake struct { + log []string + restoring []string // what each read of the recovery module's restores answers, in turn + unended []ItemRef + runFailure error + // waiting runs while the holder waits for a restore's restored; hold makes the wait last until its context ends. + waiting func() + hold bool +} + +func (f *itemBusFake) Say(event string, ref ItemRef, body map[string]any) error { + line := event + " " + ref.Module + "." + ref.Item + if o, ok := body["outcome"]; ok { + line += " " + fmt.Sprint(o) + } + if r, ok := body["restore"]; ok { + line += " restore=" + fmt.Sprint(r) + } + if e, ok := body["error"]; ok { + line += " error=" + fmt.Sprint(e) + } + if _, ok := body["at"]; !ok { + line += " NO-AT" + } + f.log = append(f.log, line) + return nil +} +func (f *itemBusFake) Unended() ([]ItemRef, error) { return f.unended, nil } +func (f *itemBusFake) Restoring(module string) (string, error) { + f.log = append(f.log, "read restores of "+module) + if len(f.restoring) == 0 { + return "", nil + } + act := f.restoring[0] + f.restoring = f.restoring[1:] + return act, nil +} +func (f *itemBusFake) WaitRestored(ctx context.Context, act string) error { + f.log = append(f.log, "restored "+act) + if f.waiting != nil { + f.waiting() + } + if f.hold { + <-ctx.Done() + return ctx.Err() + } + return nil +} + +// putOffFake is the state put-off: its keys with the time to live each was written with; expire plays the bus removing one. +type putOffFake struct { + keys map[string]PutOff + ttl map[string]time.Duration + expired map[string]func() +} + +func newPutOffFake() *putOffFake { + return &putOffFake{keys: map[string]PutOff{}, ttl: map[string]time.Duration{}, expired: map[string]func(){}} +} + +func (f *putOffFake) Hold(key string, rec PutOff, ttl time.Duration, expired func()) (func(), error) { + f.keys[key], f.ttl[key], f.expired[key] = rec, ttl, expired + return func() { delete(f.keys, key); delete(f.expired, key) }, nil +} +func (f *putOffFake) All() (map[string]PutOff, error) { return f.keys, nil } +func (f *putOffFake) expire(key string) { + cb := f.expired[key] + delete(f.keys, key) + delete(f.expired, key) + cb() +} + +// findingsFake holds the findings written and not cleared. +type findingsFake struct { + held map[string]meshconditions.Finding +} + +func (f *findingsFake) Write(_ context.Context, x meshconditions.Finding) error { + f.held[x.Scope+"."+x.ID+"."+x.Kind] = x + return nil +} +func (f *findingsFake) Clear(_ context.Context, scope, id, kind, _ string) error { + delete(f.held, scope+"."+id+"."+kind) + return nil +} +func (f *findingsFake) Held() []meshconditions.Finding { + var out []meshconditions.Finding + for _, x := range f.held { + out = append(out, x) + } + return out +} + +func itemsBackups(t *testing.T, bus *itemBusFake, broker string, runErr error) *Backups { + t.Helper() + return &Backups{Where: withData(t, composed, composedData), Node: "ace", Now: func() time.Time { return time.Date(2026, 11, 12, 3, 0, 0, 0, time.UTC) }, + Say: quiet, Bus: bus, BrokerModule: func() (string, error) { return broker, nil }, + Run: func(_ context.Context, name string, args ...string) (string, error) { + if name == "restic" && len(args) > 5 && args[5] == "backup" { + return "", runErr + } + return "[]", nil + }} +} + +func declaredOf(t *testing.T, b *Backups, module string) Declared { + t.Helper() + all, err := b.Declared() + must(t, err) + for _, d := range all { + if d.Module == module { + return d + } + } + t.Fatalf("%s is not declared", module) + return Declared{} +} + +// Design 47 G §563 (BKP4): every item of a backed-up module says item-started, then item-ended with its outcome and `at`; a +// cache item says nothing; a failed backup ends its items failed with why; the holder reads no restore for a module that does +// not hold the seat mesh-broker. +func TestEveryItemOfABackedUpModuleSaysItsStartAndItsEnd(t *testing.T) { + bus := &itemBusFake{} + b := itemsBackups(t, bus, "somebody-else", nil) + if n := b.one(context.Background(), declaredOf(t, b, "postgres"), false); !n.OK { + t.Fatalf("%+v", n) + } + want := []string{"item-started postgres.store", "item-started postgres.dumps", "item-ended postgres.store done", "item-ended postgres.dumps done"} + if !reflect.DeepEqual(bus.log, want) { + t.Fatalf("said %q", bus.log) + } + bus.log = nil + b = itemsBackups(t, bus, "somebody-else", errors.New("disk full")) + if n := b.one(context.Background(), declaredOf(t, b, "postgres"), false); n.OK { + t.Fatal("a failed backup was ok") + } + if len(bus.log) != 4 || bus.log[2] != "item-ended postgres.store failed error=disk full" { + t.Fatalf("said %q", bus.log) + } +} + +// Design 47 G §563 (BKP5): for the holder of mesh-broker the holder says its start, then reads the recovery module's restores; +// a restore that runs puts the items off naming its act, waits for that act's restored, starts again, and reads once more. +func TestTheBackupOfTheBusGivesWayToARestoreThatRuns(t *testing.T) { + bus := &itemBusFake{restoring: []string{"act-7"}} + b := itemsBackups(t, bus, "postgres", nil) + state := newPutOffFake() + b.PutOffs, b.PutOffBound = state, 45*time.Minute + // While it waits, `backed-up` lists the item put off with the restore's act and since when (BKP8, §302, §664). + bus.waiting = func() { + got, err := b.BackedUp(context.Background(), "postgres") + must(t, err) + if rec, ok := state.keys["postgres.store"]; !ok || rec.Restore != "act-7" || state.ttl["postgres.store"] != 45*time.Minute { + t.Errorf("the put-off is not a key of the state put-off with the bound as its time to live: %+v %v", state.keys, state.ttl) + } + var store *ItemReport + for i := range got[0].Data { + if got[0].Data[i].ID == "store" { + store = &got[0].Data[i] + } + } + if store == nil || store.PutOff == nil || store.PutOff.Restore != "act-7" || !store.PutOff.Since.Equal(b.Now()) { + t.Errorf("backed-up says the item as %+v", store) + } + } + if n := b.one(context.Background(), declaredOf(t, b, "postgres"), false); !n.OK { + t.Fatalf("%+v", n) + } + want := []string{ + "item-started postgres.store", "item-started postgres.dumps", "read restores of postgres", + "item-ended postgres.store put-off restore=act-7", "item-ended postgres.dumps put-off restore=act-7", + "restored act-7", + "item-started postgres.store", "item-started postgres.dumps", "read restores of postgres", + "item-ended postgres.store done", "item-ended postgres.dumps done"} + if !reflect.DeepEqual(bus.log, want) { + t.Fatalf("said %q", bus.log) + } + got, err := b.BackedUp(context.Background(), "postgres") + must(t, err) + for _, it := range got[0].Data { + if it.PutOff != nil { + t.Errorf("%s is still put off once it ran: %+v", it.ID, it.PutOff) + } + } +} + +// Design 47 G §561, §563, A-155 (BKP6): a wait for a restore past put-off-bound, the key of the state put-off expiring, writes +// the finding backup-waits for the item (scope the node), ends the wait (the backup fails, naming the bound) and holds the night's +// lock no longer; the finding is cleared when the item starts again. +func TestAWaitPastItsBoundWritesBackupWaitsAndEndsTheWait(t *testing.T) { + bus := &itemBusFake{restoring: []string{"act-7"}, hold: true} + b := itemsBackups(t, bus, "postgres", nil) + state, findings := newPutOffFake(), &findingsFake{held: map[string]meshconditions.Finding{}} + b.PutOffs, b.Findings = state, findings + bus.waiting = func() { state.expire("postgres.dumps") } + n := b.one(context.Background(), declaredOf(t, b, "postgres"), false) + if n.OK || !strings.Contains(n.Error, "put-off-bound") || !strings.Contains(n.Error, "act-7") { + t.Fatalf("the night %+v", n) + } + f, ok := findings.held["ace.postgres=2edumps.backup-waits"] + if !ok || len(findings.held) != 1 || f.Scope != "ace" || f.Kind != FindingBackupWaits || !strings.Contains(f.Summary, "act-7") { + t.Fatalf("the findings held: %+v", findings.held) + } + if len(state.keys) != 0 { + t.Errorf("the keys left: %+v", state.keys) + } + // The next night: the item starts again and the finding is cleared. + bus.restoring, bus.hold, bus.waiting = nil, false, nil + if n := b.one(context.Background(), declaredOf(t, b, "postgres"), false); !n.OK || len(findings.held) != 0 { + t.Fatalf("the night %+v, findings %+v", n, findings.held) + } +} + +// Design 47 G §561 (A-155): the setting put-off-bound is minutes in the file the manifest places; sixty when none is given. +func TestPutOffBoundReadsItsSettingFile(t *testing.T) { + file := filepath.Join(t.TempDir(), "put-off-bound") + for text, want := range map[string]time.Duration{"15\n": 15 * time.Minute, "": DefaultPutOffBound} { + must(t, os.WriteFile(file, []byte(text), 0o600)) + got, err := putOffBoundFrom(func(string) string { return file }) + if err != nil || got != want { + t.Errorf("%q gave %v, %v", text, got, err) + } + } + if got, _ := putOffBoundFrom(func(string) string { return "" }); got != DefaultPutOffBound { + t.Errorf("no file gave %v", got) + } + must(t, os.WriteFile(file, []byte("soon"), 0o600)) + if _, err := putOffBoundFrom(func(string) string { return file }); err == nil { + t.Error("a bound that is no number was taken") + } +} + +// Design 47 G §563 (BKP4): at its start the holder ends every item it said started and never ended, failed with the error stopped. +func TestItemsLeftStartedAreEndedFailedStoppedAtTheStart(t *testing.T) { + bus := &itemBusFake{unended: []ItemRef{{"postgres", "store"}}} + b := itemsBackups(t, bus, "postgres", nil) + must(t, b.StopStale()) + if !reflect.DeepEqual(bus.log, []string{"item-ended postgres.store failed error=stopped"}) { + t.Fatalf("said %q", bus.log) + } +} diff --git a/modules/restic/cmd/restic-backups/live_integration_test.go b/modules/restic/cmd/restic-backups/live_integration_test.go new file mode 100644 index 000000000..87d16daae --- /dev/null +++ b/modules/restic/cmd/restic-backups/live_integration_test.go @@ -0,0 +1,244 @@ +//go:build integration + +package main + +// The module's live path on a bus of this test alone (design 47 G §561, §563, A-150, A-155): the night's own broker account. + +import ( + "context" + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/testbus" +) + +// A put-off is a key of the state put-off with a time to live of its own: it is listed while it lasts, and the bus removing it +// at the end of its time is heard as the expiry, once; a key released by the holder is heard as nothing. +func TestAPutOffKeyIsListedAndItsExpiryIsHeardFromTheBus(t *testing.T) { + url := testbus.URL(t) + nc, err := nats.Connect(url) + if err != nil { + t.Fatal(err) + } + defer nc.Close() + js, _ := jetstream.New(nc) + ctx, cancel := context.WithTimeout(context.Background(), time.Minute) + defer cancel() + if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "restic_" + PutOffStateName, LimitMarkerTTL: time.Second}); err != nil { + t.Fatal(err) + } + conn, err := process.Connect(process.Account{URL: url, Node: "ace", Module: "restic", User: "ace.restic", Password: "unused"}) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + st := stateOfPutOffs{conn.State(PutOffStateName)} + now := time.Date(2026, 11, 12, 3, 0, 0, 0, time.UTC) + expired := make(chan string, 4) + if _, err := st.Hold("postgres.store", PutOff{Restore: "act-7", Since: now}, 2*time.Second, func() { expired <- "store" }); err != nil { + t.Fatal(err) + } + release, err := st.Hold("postgres.dumps", PutOff{Restore: "act-7", Since: now}, time.Hour, func() { expired <- "dumps" }) + if err != nil { + t.Fatal(err) + } + all, err := st.All() + if err != nil || len(all) != 2 || all["postgres.store"].Restore != "act-7" || !all["postgres.store"].Since.Equal(now) { + t.Fatalf("the keys %+v, %v", all, err) + } + select { + case who := <-expired: + if who != "store" { + t.Fatalf("%s expired", who) + } + case <-time.After(30 * time.Second): + t.Fatal("the bus did not remove the key at the end of its time") + } + release() + for { + all, err = st.All() + if err != nil { + t.Fatal(err) + } + if len(all) == 0 { + break + } + select { + case who := <-expired: + t.Fatalf("%s expired after its release", who) + case <-time.After(100 * time.Millisecond): + } + } +} + +// The night as the node starts it: its credential is read from where the manifest puts it (the process night's MESH_BROKER_FILE, +// the module's own secret broker), it says each item started and ended on the seat node-backup, gives way to a restore of the +// recovery module that runs, puts its items off as keys of the state put-off and starts again when the restore says restored; a +// restore that does not end within put-off-bound expires its keys on the bus, writes backup-waits into the module's findings and +// ends the wait. The streams are the ones the bus composer makes. +func TestTheNightSaysItsItemsAndGivesWayToARestoreOnTheBus(t *testing.T) { + // The manifest's own place of the credential, and the environment the night is given. + raw, err := os.ReadFile(filepath.Join(moduleDir, "module.json")) + if err != nil { + t.Fatal(err) + } + var m struct { + OwnSecrets map[string]string `json:"own-secrets"` + Resources []struct { + ID string `json:"id"` + Env map[string]string `json:"env"` + } `json:"resources"` + } + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatal(err) + } + dir := t.TempDir() + env := map[string]string{} + for _, r := range m.Resources { + if r.ID == "night" { + for k, v := range r.Env { + env[k] = strings.ReplaceAll(v, "${dir:state}", dir) + } + } + } + if m.OwnSecrets["broker"] != "${dir:state}/broker" || env[process.BrokerFileEnv] != filepath.Join(dir, "broker") { + t.Fatalf("the manifest places the broker file at %q and gives the night %q", m.OwnSecrets["broker"], env[process.BrokerFileEnv]) + } + url := testbus.URL(t) + cred, _ := json.Marshal(process.Account{URL: url, Node: "ace", Module: "restic", User: "ace.restic", Password: "unused"}) + if err := os.WriteFile(env[process.BrokerFileEnv], cred, 0o600); err != nil { + t.Fatal(err) + } + live, err := connectItems(func(k string) string { return env[k] }) + if err != nil || live == nil { + t.Fatalf("the night's credential at its place was not used: %v", err) + } + defer live.Close() + + nc, err := nats.Connect(url) + if err != nil { + t.Fatal(err) + } + defer nc.Close() + js, _ := jetstream.New(nc) + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + for _, b := range []jetstream.KeyValueConfig{{Bucket: "restic_" + PutOffStateName, LimitMarkerTTL: time.Second}, {Bucket: "restic_findings"}} { + if _, err := js.CreateKeyValue(ctx, b); err != nil { + t.Fatal(err) + } + } + for stream, subject := range map[string]string{ + streamItemStarted: "mesh.seat.node-backup.event.item-started.>", streamItemEnded: "mesh.seat.node-backup.event.item-ended.>", + streamRestoreStarted: "mesh.seat.mesh-recovery.event.restore-started.>", streamRestored: "mesh.seat.mesh-recovery.event.restored.>"} { + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{subject}, MaxMsgsPerSubject: 1}); err != nil { + t.Fatal(err) + } + } + for _, s := range []jetstream.StreamConfig{{Name: "RESTIC_FINDINGS", Subjects: []string{"mesh.stream.restic.event.>"}}, + {Name: "FINDINGS_ASKED", Subjects: []string{meshconditions.SubjectFindingsAsked}}} { + if _, err := js.CreateStream(ctx, s); err != nil { + t.Fatal(err) + } + } + // The assigner's answer to who holds the seat mesh-broker: the module postgres. + if _, err := nc.Subscribe("mesh.seat.mesh-module-assigner.call.assignments.*", func(msg *nats.Msg) { + body, _ := json.Marshal(map[string]any{"result": map[string]any{"seats": []any{map[string]any{"seat": BrokerSeat, + "holders": []any{map[string]any{"module": "postgres"}}}}}, "node": "ace"}) + _ = msg.Respond(body) + }); err != nil { + t.Fatal(err) + } + findings, closeFindings, err := live.Findings(ctx, t.Logf) + if err != nil { + t.Fatal(err) + } + defer closeFindings() + + b := &Backups{Where: withData(t, composed, composedData), Node: "ace", Now: time.Now, Say: quiet, Bus: live, + BrokerModule: live.Broker, PutOffs: live.PutOffs(), Findings: findings, + Run: func(context.Context, string, ...string) (string, error) { return "[]", nil }} + say := func(event, act string) { + body, _ := json.Marshal(map[string]any{"act": act, "at": time.Now().UTC()}) + if _, err := js.Publish(ctx, "mesh.seat.mesh-recovery.event."+event+".postgres.store", body); err != nil { + t.Fatal(err) + } + } + lastEnded := func(item string) string { + msg, err := js.Conn().RequestWithContext(ctx, "$JS.API.STREAM.MSG.GET."+streamItemEnded, []byte(`{"last_by_subj":"`+ + itemEventSubject(EventItemEnded, "ace", "postgres", item)+`"}`)) + if err != nil { + t.Fatal(err) + } + var got struct { + Message struct { + Data []byte `json:"data"` + } `json:"message"` + } + if err := json.Unmarshal(msg.Data, &got); err != nil { + t.Fatal(err) + } + return string(got.Message.Data) + } + + // A restore runs and ends: the items are put off as keys, then start again and end done. + say("restore-started", "act-7") + kv, _ := js.KeyValue(ctx, "restic_"+PutOffStateName) + watch, err := kv.Watch(ctx, "postgres.store", jetstream.UpdatesOnly()) + if err != nil { + t.Fatal(err) + } + ended := make(chan Night, 1) + go func() { ended <- b.one(ctx, declaredOf(t, b, "postgres"), false) }() + select { + case e := <-watch.Updates(): + if e == nil || e.Operation() != jetstream.KeyValuePut { + t.Fatalf("the put-off key came as %+v", e) + } + case <-time.After(30 * time.Second): + t.Fatal("no put-off key was written") + } + say("restored", "act-7") + if n := <-ended; !n.OK { + t.Fatalf("the night after the restore: %+v", n) + } + left, err := live.Unended() + if err != nil || len(left) != 0 { + t.Fatalf("items left started: %+v, %v", left, err) + } + if got := lastEnded("store"); !strings.Contains(got, `"outcome":"done"`) { + t.Fatalf("the last end of the item store: %s", got) + } + + // A restore that does not end: the bound (a second) passes, the bus removes the keys, backup-waits is written. + b.PutOffBound = time.Second + say("restore-started", "act-8") + began := time.Now() + n := b.one(ctx, declaredOf(t, b, "postgres"), false) + if n.OK || !strings.Contains(n.Error, "put-off-bound") { + t.Fatalf("the night past the bound: %+v", n) + } + if took := time.Since(began); took > 20*time.Second { + t.Fatalf("the wait ended %v after it began: the bound's expiry did not end it", took) + } + fkv, _ := js.KeyValue(ctx, "restic_findings") + ks, err := fkv.Keys(ctx) + waits := 0 + for _, k := range ks { + if strings.HasSuffix(k, "."+FindingBackupWaits) && strings.HasPrefix(k, "ace.postgres=2e") { + waits++ + } + } + if err != nil || waits != 1 { + t.Fatalf("the findings kept: %v, %v", ks, err) + } +} diff --git a/modules/restic/cmd/restic-backups/main.go b/modules/restic/cmd/restic-backups/main.go index 14579bd33..a117dd6ab 100644 --- a/modules/restic/cmd/restic-backups/main.go +++ b/modules/restic/cmd/restic-backups/main.go @@ -47,6 +47,39 @@ func main() { if len(os.Args) > 1 { way = os.Args[1] } + // The night and the tools' runtime each say their items on the bus as the module's own account, and read the recovery + // module's restores (design 47 G §563, A-150); a process given no credential says none. + if way == "night" || way == "" { + live, err := connectItems(os.Getenv) + if err != nil { + say("the bus cannot be reached with the module's own credential: %v", err) + os.Exit(1) + } + if live != nil { + defer live.Close() + b.Bus, b.BrokerModule, b.PutOffs = live, live.Broker, live.PutOffs() + bound, err := putOffBoundFrom(os.Getenv) + if err != nil { + say("%v", err) + os.Exit(1) + } + b.PutOffBound = bound + if way == "night" { + findings, closeFindings, err := live.Findings(context.Background(), say) + if err != nil { + say("the finding writer cannot be started: %v", err) + os.Exit(1) + } + defer closeFindings() + b.Findings = findings + } + } + // Every item this node said started and never ended is ended failed, stopped (§563). + if err := b.StopStale(); err != nil { + say("%v", err) + os.Exit(1) + } + } if way == "" { b.Emit = func(event string, body any) error { return stdio.Emit(event, body) } b.Readings = busReadings(b.Node) diff --git a/modules/restic/cmd/restic-backups/putoff.go b/modules/restic/cmd/restic-backups/putoff.go new file mode 100644 index 000000000..776b617c2 --- /dev/null +++ b/modules/restic/cmd/restic-backups/putoff.go @@ -0,0 +1,185 @@ +package main + +// An item put off for a restore (design 47 G §563, §302, §664, A-155): `backed-up` lists it with the restore's act and since +// when. The wait is a bound key `.` in the module's own state `put-off`, with a time to live of its own (the setting +// put-off-bound): the key is the record `backed-up` reads, its expiry is heard as the bus's own change, raises the finding +// `backup-waits` and ends the wait. No file and no timer keeps it. + +import ( + "context" + "errors" + "fmt" + "os" + "strconv" + "strings" + "time" + + "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + "git.novox.be/novox/mesh-sdk/go/keys" +) + +// PutOff says an item gave way to a restore: the act of that restore and since when. +type PutOff struct { + Restore string `json:"restore"` + Since time.Time `json:"since"` +} + +// FindingBackupWaits is the backup holder's finding of an item put off past its bound (§561). +const FindingBackupWaits = "backup-waits" + +// DefaultPutOffBound is the setting put-off-bound's default (§561, §557): sixty minutes. +const DefaultPutOffBound = 60 * time.Minute + +// PutOffState is the module's own state `put-off`, whose keys live a time of their own. +type PutOffState interface { + // Hold writes the key with its record and time to live, watched before it is written; expired is called once when the bus + // removes the key at the end of its time. release removes the key and ends the watch, calling nothing. + Hold(key string, rec PutOff, ttl time.Duration, expired func()) (release func(), err error) + // All reads every key with its record. + All() (map[string]PutOff, error) +} + +// Findings is the backup holder's finding writer (mesh-sdk's held writer). +type Findings interface { + Write(ctx context.Context, f meshconditions.Finding) error + Clear(ctx context.Context, scope, id, kind, why string) error + Held() []meshconditions.Finding +} + +func putOffKey(ref ItemRef) string { + return keys.EscapeName(ref.Module) + "." + keys.EscapeName(ref.Item) +} + +func putOffRef(key string) (ItemRef, bool) { + m, i, ok := strings.Cut(key, ".") + if !ok { + return ItemRef{}, false + } + module, e1 := keys.UnescapeName(m) + item, e2 := keys.UnescapeName(i) + return ItemRef{Module: module, Item: item}, e1 == nil && e2 == nil +} + +func (b *Backups) putOffBound() time.Duration { + if b.PutOffBound > 0 { + return b.PutOffBound + } + return DefaultPutOffBound +} + +// errPutOffExpired ends the wait for a restore past the bound. +var errPutOffExpired = errors.New("the restore did not end within the bound put-off-bound") + +// keepPutOff records a module's items as put off for a restore, since now, each a bound key; expired is called once when the +// first of them is removed by the bus at the end of its time. It answers how to let the keys go. +func (b *Backups) keepPutOff(items []ItemRef, act string, expired func(ItemRef)) (func(), error) { + if b.PutOffs == nil { + return func() {}, nil + } + var releases []func() + release := func() { + for _, r := range releases { + r() + } + } + for _, ref := range items { + ref := ref + r, err := b.PutOffs.Hold(putOffKey(ref), PutOff{Restore: act, Since: b.Now().UTC()}, b.putOffBound(), func() { expired(ref) }) + if err != nil { + release() + return nil, fmt.Errorf("the put-off of %s.%s could not be kept: %w", ref.Module, ref.Item, err) + } + releases = append(releases, r) + } + return release, nil +} + +// writeWaits says the finding backup-waits for an item put off past its bound, naming the restore's act (§561). +func (b *Backups) writeWaits(ref ItemRef, act string) { + if b.Findings == nil { + return + } + f := meshconditions.Finding{Scope: b.Node, ID: keys.EscapeName(ref.Module + "." + ref.Item), Kind: FindingBackupWaits, + Subject: meshconditions.Subject{Scope: "module", ID: "restic"}, + Severity: meshconditions.FindingSeverityWarning, Headline: "A backup waits on a restore", + Summary: fmt.Sprintf("the backup of %s.%s was put off for the restore %s of the recovery module and that restore has not ended within %s", ref.Module, ref.Item, act, b.putOffBound()), + Explanation: fmt.Sprintf("the item %s.%s gave way to the restore %s at its start and waits for its restored", ref.Module, ref.Item, act), + Needs: "the restore ended or cancelled in the recovery module (recovery_cancel); the backup then starts again at the next night", + Source: "backups", Observer: "restic", At: b.Now().UTC()} + if err := b.Findings.Write(context.Background(), f); err != nil { + b.Say("the finding %s of %s.%s could not be written: %v", FindingBackupWaits, ref.Module, ref.Item, err) + } +} + +// clearWaits clears the finding backup-waits of the items that started again. +func (b *Backups) clearWaits(items []ItemRef) { + if b.Findings == nil { + return + } + held := map[string]bool{} + for _, f := range b.Findings.Held() { + if f.Kind == FindingBackupWaits { + held[f.Scope+"\x00"+f.ID] = true + } + } + for _, ref := range items { + id := keys.EscapeName(ref.Module + "." + ref.Item) + if !held[b.Node+"\x00"+id] { + continue + } + if err := b.Findings.Clear(context.Background(), b.Node, id, FindingBackupWaits, "the item started again"); err != nil { + b.Say("the finding %s of %s.%s could not be cleared: %v", FindingBackupWaits, ref.Module, ref.Item, err) + } + } +} + +// withPutOff says on a module's item reports which items are put off, read from the state's keys. +func (b *Backups) withPutOff(module string, reports []ItemReport) []ItemReport { + if b.PutOffs == nil { + return reports + } + all, err := b.PutOffs.All() + if err != nil { + b.Say("the items put off could not be read: %v", err) + return reports + } + for key, rec := range all { + ref, ok := putOffRef(key) + if !ok || ref.Module != module { + continue + } + for i := range reports { + if reports[i].ID == ref.Item { + put := rec + reports[i].PutOff = &put + } + } + } + return reports +} + +// PutOffBoundEnv names the file of the setting put-off-bound, in minutes: the time the wait for a restore may last before +// backup-waits is written and the wait ends (§561, §557). +const PutOffBoundEnv = "MESH_BACKUP_PUT_OFF_BOUND_FILE" + +// putOffBoundFrom reads the setting from its file; it is the default (sixty minutes) when the process is given no file or an +// empty one, and refused when it does not read as a number of minutes above zero. +func putOffBoundFrom(getenv func(string) string) (time.Duration, error) { + path := strings.TrimSpace(getenv(PutOffBoundEnv)) + if path == "" { + return DefaultPutOffBound, nil + } + raw, err := os.ReadFile(path) + if err != nil { + return 0, fmt.Errorf("the setting put-off-bound cannot be read: %w", err) + } + text := strings.TrimSpace(string(raw)) + if text == "" { + return DefaultPutOffBound, nil + } + minutes, err := strconv.ParseFloat(text, 64) + if err != nil || minutes <= 0 { + return 0, fmt.Errorf("the setting put-off-bound is %q, not a number of minutes above zero", text) + } + return time.Duration(minutes * float64(time.Minute)), nil +} diff --git a/modules/restic/cmd/restic-backups/templates_test.go b/modules/restic/cmd/restic-backups/templates_test.go index 7d0718b57..e0c416421 100644 --- a/modules/restic/cmd/restic-backups/templates_test.go +++ b/modules/restic/cmd/restic-backups/templates_test.go @@ -4,6 +4,7 @@ import ( "encoding/json" "os" "path/filepath" + "strings" "testing" ) @@ -62,3 +63,58 @@ func TestEveryFileTakesItsTextFromATemplate(t *testing.T) { t.Errorf("the build packs files/ into %d archives, not one", packs) } } + +// Design 47 G §561, A-155: the put-off wait is a bound key of a state of the module's own with a key-ttl, the finding backup-waits +// goes through the findings template and its events, the setting put-off-bound is a file the night reads, and the module hears +// the findings asked. +func TestThePutOffBoundAndItsStatesAreInTheManifest(t *testing.T) { + raw, err := os.ReadFile(filepath.Join(moduleDir, "module.json")) + if err != nil { + t.Fatal(err) + } + var m struct { + Consumes []string `json:"consumes"` + Requires []struct { + Name, Template string + Options map[string]any + } `json:"requires"` + Config map[string]map[string]any `json:"config"` + Resources []map[string]any `json:"resources"` + } + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatal(err) + } + got := map[string]bool{} + for _, r := range m.Requires { + got[r.Name+"/"+r.Template] = true + if r.Name == "put-off" && r.Options["key-ttl"] != true { + t.Errorf("the state put-off has no key-ttl: %v", r.Options) + } + } + for _, want := range []string{"put-off/state", "findings/state", "finding-events/events"} { + if !got[want] { + t.Errorf("the manifest requires no %s: %v", want, got) + } + } + if m.Config["put-off-bound"]["default"] != float64(60) { + t.Errorf("the setting put-off-bound is %v", m.Config["put-off-bound"]) + } + heard := false + for _, c := range m.Consumes { + heard = heard || c == "mesh-conditions.findings-asked" + } + if !heard { + t.Error("the module does not hear mesh-conditions.findings-asked") + } + for _, r := range m.Resources { + if r["id"] == "night" { + env, _ := r["env"].(map[string]any) + if env[PutOffBoundEnv] != "${dir:state}/put-off-bound" { + t.Errorf("the night is not given the file of put-off-bound: %v", env) + } + } + if r["id"] == "put-off-bound" && strings.TrimSpace(templateText(t, r)) != "${config:put-off-bound}" { + t.Errorf("the file of put-off-bound does not hold its setting") + } + } +} diff --git a/modules/restic/files/put-off-bound b/modules/restic/files/put-off-bound new file mode 100644 index 000000000..c33d7f57e --- /dev/null +++ b/modules/restic/files/put-off-bound @@ -0,0 +1 @@ +${config:put-off-bound} diff --git a/modules/restic/go.mod b/modules/restic/go.mod index 84ee5a573..fe82d5eb6 100644 --- a/modules/restic/go.mod +++ b/modules/restic/go.mod @@ -2,11 +2,14 @@ module restic go 1.26.0 -require git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000 +require ( + git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000 +) require ( github.com/klauspost/compress v1.20.1 // indirect - github.com/nats-io/nats.go v1.54.0 // indirect + github.com/nats-io/nats.go v1.54.0 github.com/nats-io/nkeys v0.4.16 // indirect github.com/nats-io/nuid v1.0.1 // indirect golang.org/x/crypto v0.57.0 // indirect diff --git a/modules/restic/module.json b/modules/restic/module.json index 462ff7aec..6482eb39e 100644 --- a/modules/restic/module.json +++ b/modules/restic/module.json @@ -16,6 +16,11 @@ ] } ], + "consumes": [ + "mesh-recovery.restore-started", + "mesh-recovery.restored", + "mesh-conditions.findings-asked" + ], "emits": [ "backed-up" ], @@ -25,7 +30,8 @@ "backup_deleted" ], "own-secrets": { - "repository": "${dir:state}/repository.secret" + "repository": "${dir:state}/repository.secret", + "broker": "${dir:state}/broker" }, "data": { "own": [ @@ -76,6 +82,14 @@ "maximum": 9999, "trusted": true, "default": 7 + }, + "put-off-bound": { + "type": "number", + "description": "minutes an item put off for a restore of the recovery module is waited for before the finding backup-waits is written and the wait ends, so the night never holds its lock without a bound; timed by a bound key of the state put-off (design 47 G sections 561, 557, A-155).", + "minimum": 1, + "maximum": 10080, + "trusted": false, + "default": 60 } }, "resources": [ @@ -114,6 +128,13 @@ "trusted": true, "template": "files/retention.conf" }, + { + "id": "put-off-bound", + "type": "file", + "path": "${dir:state}/put-off-bound", + "mode": "0600", + "template": "files/put-off-bound" + }, { "id": "tool", "type": "package", @@ -139,7 +160,9 @@ "MESH_BACKUP_REPOSITORY": "${dir:repository}", "MESH_BACKUP_PASSWORD_FILE": "${dir:state}/repository.secret", "MESH_BACKUP_STATE": "${dir:state}", - "MESH_BACKUP_DATA": "${dir:state}/data.conf" + "MESH_BACKUP_DATA": "${dir:state}/data.conf", + "MESH_BROKER_FILE": "${dir:state}/broker", + "MESH_BACKUP_PUT_OFF_BOUND_FILE": "${dir:state}/put-off-bound" }, "health": { "kind": "unit" @@ -160,7 +183,9 @@ "MESH_BACKUP_REPOSITORY": "${dir:repository}", "MESH_BACKUP_PASSWORD_FILE": "${dir:state}/repository.secret", "MESH_BACKUP_STATE": "${dir:state}", - "MESH_BACKUP_DATA": "${dir:state}/data.conf" + "MESH_BACKUP_DATA": "${dir:state}/data.conf", + "MESH_BROKER_FILE": "${dir:state}/broker", + "MESH_BACKUP_PUT_OFF_BOUND_FILE": "${dir:state}/put-off-bound" }, "health": { "kind": "tool", @@ -189,7 +214,9 @@ "MESH_BACKUP_REPOSITORY": "${dir:repository}", "MESH_BACKUP_PASSWORD_FILE": "${dir:state}/repository.secret", "MESH_BACKUP_STATE": "${dir:state}", - "MESH_BACKUP_DATA": "${dir:state}/data.conf" + "MESH_BACKUP_DATA": "${dir:state}/data.conf", + "MESH_BROKER_FILE": "${dir:state}/broker", + "MESH_BACKUP_PUT_OFF_BOUND_FILE": "${dir:state}/put-off-bound" } }, { @@ -204,5 +231,34 @@ ], "state": [ "good" + ], + "requires": [ + { + "name": "put-off", + "storage": "key-value", + "template": "state", + "options": { + "key-ttl": true + } + }, + { + "name": "findings", + "storage": "key-value", + "template": "state" + }, + { + "name": "finding-events", + "storage": "stream", + "template": "events", + "options": { + "events": [ + "finding-written.*", + "finding-cleared.*", + "findings-stated.*" + ], + "max-age-days": 7, + "max-mib": 16 + } + } ] }