Merge branch 'feat/47-G-fix11-c2' into feat/47-G-fix11-b7

This commit is contained in:
2026-11-12 05:43:07 +01:00
41 changed files with 1894 additions and 139 deletions
+4 -4
View File
@@ -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
@@ -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")
}
}
+34 -17
View File
@@ -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 `<owner>/<name>`, 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
@@ -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`.
@@ -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"
}
}
@@ -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"
}
}
+3 -4
View File
@@ -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",
@@ -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
@@ -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))
}
@@ -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)
}},
}
}
@@ -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)
}
}
+3 -1
View File
@@ -349,7 +349,9 @@
"gate_record",
"gate_record_end",
"gate_record_repair",
"gate_records"
"gate_records",
"gate_source",
"gate_source_again"
],
"config": {
"record-retention": {
@@ -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:
@@ -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.
+2 -2
View File
@@ -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"
@@ -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})
@@ -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
}
@@ -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.>",
} {
@@ -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-<node> (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")
}
}
+64 -2
View File
@@ -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 `<asker>.<id>` 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.
@@ -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
+20
View File
@@ -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
@@ -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")
}
}
+1
View File
@@ -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"]}
}
+9 -1
View File
@@ -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"
+1 -1
View File
@@ -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`.
+12 -9
View File
@@ -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 `<owner>/<name>`
// (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, `<owner>/<name>`.
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 {
@@ -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 `<owner>/<name>`.
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)
}
}
}
+1 -1
View File
@@ -9,7 +9,7 @@
"git": "${dir:mesh-state}/git.json"
},
"consumes": [
"gitea.pull.merged"
"git.pull-merged"
],
"tools": [
"records_search",
+26 -4
View File
@@ -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
}
@@ -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'"\\$` + "`" + `]*$`)
@@ -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
}
+212
View File
@@ -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
})
}
@@ -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)
}
}
@@ -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)
}
}
+33
View File
@@ -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)
+185
View File
@@ -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 `<module>.<item>` 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
}
@@ -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")
}
}
}
+1
View File
@@ -0,0 +1 @@
${config:put-off-bound}
+5 -2
View File
@@ -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
+60 -4
View File
@@ -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
}
}
]
}