diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go index 74f682a7e..1fd9bcaa4 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go @@ -17,6 +17,7 @@ import ( "sync" "time" + "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" gitseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/git" stdio "git.novox.be/novox/mesh-sdk/go" ) @@ -466,6 +467,7 @@ func tools(h *Holder, l *listening) []stdio.Tool { return out, nil }}, } + list = append(list, recordTools(h, meshhandactsclient.Recorder{Ask: stdio.Ask})...) // A verb that changes a delivery asks for a look at what it makes possible (react.go). for i := range list { run := list[i].Run diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/manifest_test.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/manifest_test.go index 3c35d1126..9d0edb064 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/manifest_test.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/manifest_test.go @@ -115,7 +115,7 @@ func TestItClaimsTheDeliverySeatAndFollowsWhatItHandles(t *testing.T) { // merge state through the one statuses tool the seat does not carry yet (seatForge.mergeState); the forge's writes are the // bridge's (ADR 0437 §8). It uses operator-channel only to say its conditions (ADR 0437 §9, issue 1275): the teller asks // (ADR 0437 §6). - want := []string{"seat:git.pull", "seat:git.statuses", "seat:git.open-pulls", "seat:git.contains", mergeStateTool} + want := []string{"seat:git.pull", "seat:git.statuses", "seat:git.open-pulls", "seat:git.contains", mergeStateTool, "seat:mesh-hand-acts.hand-act"} if !reflect.DeepEqual(m.Uses, []string{"operator-channel"}) { t.Fatalf("uses %v: operator-channel, for its conditions alone", m.Uses) } diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record.go index 0c0934532..4bf027ce0 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record.go @@ -271,6 +271,8 @@ type RecordLedger interface { Apply(instance, operation string, body any) (uint64, error) // Get is the instance's state and the sequence of its last operation; a nil state when it has none. Get(instance string) (json.RawMessage, uint64, error) + // Remove removes a record with its operations and its snapshot (OpRemove); an error when any of them stays. + Remove(instance string) error } // sdkLedger is the gate's ledger through the runtime: the tool runner's ledger calls, on the gate's own ledger only @@ -281,23 +283,42 @@ type sdkLedger struct{} var ledgerCall = stdio.BusState func (sdkLedger) Apply(instance, operation string, body any) (uint64, error) { + got, err := applyCall(instance, operation, body) + return got.Sequence, err +} + +// applyCall is one operation applied through the runtime, and the ledger's answer. +func applyCall(instance, operation string, body any) (buscalls.Applied, error) { + var got buscalls.Applied raw, err := json.Marshal(body) if err != nil { - return 0, err + return got, err } out, err := ledgerCall(buscalls.LedgerApply, buscalls.ApplyRequest{Ledger: RecordsLedger, Instance: instance, Operation: operation, Body: raw}) if err != nil { - return 0, err + return got, err } - var got buscalls.Applied if err := json.Unmarshal(out, &got); err != nil { - return 0, fmt.Errorf("the ledger's answer to %s on %s is not readable: %w", operation, instance, err) + return got, fmt.Errorf("the ledger's answer to %s on %s is not readable: %w", operation, instance, err) } if !got.Taken { - return 0, fmt.Errorf("the ledger did not take %s on %s", operation, instance) + return got, fmt.Errorf("the ledger did not take %s on %s", operation, instance) } - return got.Sequence, nil + return got, nil +} + +// Remove applies OpRemove and reads the ledger's answer: a removal that left the record's operations, its snapshot or its +// state is an error, so the retention says it rather than leaving a record's history to grow unseen. +func (sdkLedger) Remove(instance string) error { + got, err := applyCall(instance, OpRemove, map[string]any{}) + if err != nil { + return err + } + if !got.Removed { + return fmt.Errorf("the ledger removed %s but kept its operations or its snapshot (removed false at sequence %d)", instance, got.Sequence) + } + return nil } func (sdkLedger) Get(instance string) (json.RawMessage, uint64, error) { @@ -540,7 +561,7 @@ func (rs *Records) Sweep(now time.Time) error { if err := rs.sayFreedAll(r, "the record is past its retention"); err != nil { return err } - if _, err := rs.Ledger.Apply(e.Instance, OpRemove, map[string]any{}); err != nil { + if err := rs.Ledger.Remove(e.Instance); err != nil { return err } rs.forget(e.Instance) diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record_test.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record_test.go index 9e141b073..e9fb985e6 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record_test.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record_test.go @@ -2,6 +2,7 @@ package main import ( "encoding/json" + "errors" "fmt" "os" "path/filepath" @@ -11,9 +12,11 @@ import ( "testing" "time" + "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" mdc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient" "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/buscalls" "git.novox.be/novox/mesh-sdk/go/catalogue" ) @@ -133,6 +136,11 @@ func (f *ledgerFake) Apply(instance, operation string, body any) (uint64, error) return f.seq[instance], nil } +func (f *ledgerFake) Remove(instance string) error { + _, err := f.Apply(instance, OpRemove, map[string]any{}) + return err +} + func (f *ledgerFake) Get(instance string) (json.RawMessage, uint64, error) { s := f.states[instance] if s == nil || string(s) == "null" { @@ -618,3 +626,98 @@ func TestACandidatesHeadsReplacedOrClosedAreSaid(t *testing.T) { } }) } + +// handActsFake is the hand-act log: what it took, or a refusal. +type handActsFake struct { + took []meshhandactsclient.Request + refuse error +} + +func (f *handActsFake) Record(q meshhandactsclient.Request) (meshhandactsclient.Taken, error) { + if f.refuse != nil { + return meshhandactsclient.Taken{}, f.refuse + } + f.took = append(f.took, q) + return meshhandactsclient.Taken{Log: "hand-acts"}, nil +} + +// W11.13 (design 47 *The delivery record*, its tools): gate_record reads a record whole, gate_records lists them with +// their fill, gate_record_end ends one by hand after the hand-act log took it; and a removal that keeps the record's +// operations is said, never taken quietly (buscalls.Applied.Removed). +func TestTheRecordsToolsReadThemAndEndOneByHand(t *testing.T) { + at := time.Date(2026, 11, 9, 22, 0, 0, 0, time.UTC) + heads := map[string]string{"novox/app@111111111111": strings.Repeat("1", 40)} + setup := func(t *testing.T) (*Records, *Record, *[]string) { + var said []string + rs := &Records{Ledger: newLedgerFake(t), Now: func() time.Time { return at }, + Say: func(subject string, body any, id string) error { said = append(said, subject+" "+id); return nil }} + r, _, err := rs.Open("47/G", heads, false) + if err != nil { + t.Fatal(err) + } + if err := rs.AddEntry(r, RecordEntry{Module: "mesh-x", Fingerprint: "src2:f1", Node: "ace"}); err != nil { + t.Fatal(err) + } + if err := rs.SetTrace(r, RecordTrace{To: "assignment-9"}); err != nil { + t.Fatal(err) + } + return rs, r, &said + } + t.Run("gate_record and gate_records read a record whole and its fill", func(t *testing.T) { + rs, r, _ := setup(t) + got, err := rs.RecordOf(r.Instance, "") + if err != nil || len(got) != 1 || got[0].Record.Milestone != "47/G" || got[0].Waits != "waiting for mesh-x on ace" || + got[0].Bytes == 0 || got[0].Of != 256<<10 { + t.Fatalf("%+v %v", got, err) + } + if byMilestone, err := rs.RecordOf("", "47/G"); err != nil || len(byMilestone) != 1 { + t.Fatalf("by milestone %+v %v", byMilestone, err) + } + if _, err := rs.RecordOf("", "46/F"); err == nil { + t.Fatal("a milestone with no record answers nothing") + } + lines, err := rs.Lines() + if err != nil || lines["open"] != 1 || len(lines["records"].([]recordLine)) != 1 || lines["records"].([]recordLine)[0].Entries != 1 { + t.Fatalf("%+v %v", lines, err) + } + }) + t.Run("gate_record_end ends a record only once the hand-act log took it", func(t *testing.T) { + rs, r, said := setup(t) + log := &handActsFake{refuse: errors.New("no responders")} + if _, err := rs.EndByHand(r.Instance, "", "jochen@shanks", log); err == nil { + t.Fatal("ended with no reason") + } + if _, err := rs.EndByHand(r.Instance, "g14 is gone for good", "jochen@shanks", log); err == nil { + t.Fatal("ended though the hand-act log refused it") + } + if back, _ := rs.Read(r.Instance); back.Ended != nil { + t.Fatal("a record is ended while no hand act was recorded") + } + log.refuse = nil + v, err := rs.EndByHand(r.Instance, "g14 is gone for good", "jochen@shanks", log) + if err != nil || v.Record.Ended == nil || v.Record.Ended.By != "jochen@shanks" || len(log.took) != 1 { + t.Fatalf("%+v %v (log %+v)", v, err, log.took) + } + if !slices.Contains(*said, meshdelivery.SubjectRecordEnded+" record-ended."+r.Instance) { + t.Fatalf("said %v", *said) + } + if _, err := rs.EndByHand(r.Instance, "again", "jochen@shanks", log); err == nil || len(log.took) != 1 { + t.Fatal("a record ended is ended again") + } + }) + t.Run("a removal that keeps the record's operations is an error", func(t *testing.T) { + was := ledgerCall + t.Cleanup(func() { ledgerCall = was }) + removed := false + ledgerCall = func(method string, body any) (json.RawMessage, error) { + return json.Marshal(buscalls.Applied{Taken: true, Sequence: 4, Removed: removed}) + } + if err := (sdkLedger{}).Remove("record/47_G/0011223344556677"); err == nil || !strings.Contains(err.Error(), "kept its operations") { + t.Fatalf("a removal that kept its operations: %v", err) + } + removed = true + if err := (sdkLedger{}).Remove("record/47_G/0011223344556677"); err != nil { + t.Fatal(err) + } + }) +} diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/recordtools.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/recordtools.go new file mode 100644 index 000000000..72faf22d1 --- /dev/null +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/recordtools.go @@ -0,0 +1,193 @@ +package main + +// The delivery record's tools (design 47 *The delivery record*, its tools against the four lines a module's tools answer): +// gate_record reads one record whole, gate_records lists them with their fill, gate_consumer reads the record's consumer, +// and gate_record_end ends a record left open, recorded first in the hand-act log of the seat mesh-hand-acts. + +import ( + "encoding/json" + "errors" + "fmt" + "strings" + + "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// valueLimit is the bus's limit of one value, which a record's state is held under (design 47: a record of 30 modules on +// four nodes is about 30 KiB of it). +const valueLimit = 256 << 10 + +// recordView is one record as gate_record answers it: its state, the sequence of its last operation, its size against +// the value limit, and while open why it has not ended. +type recordView struct { + Instance string `json:"instance"` + Sequence uint64 `json:"sequence"` + Bytes int `json:"bytes"` + Of int `json:"of-bytes"` + Waits string `json:"waits,omitempty"` + Record *Record `json:"record"` +} + +func viewOf(r *Record) recordView { + raw, _ := json.Marshal(r) + v := recordView{Instance: r.Instance, Sequence: r.Sequence, Bytes: len(raw), Of: valueLimit, Record: r} + if r.Ended == nil { + v.Waits = allRun(r) + } + return v +} + +// RecordOf is one record whole, by its instance, or every record of a milestone. +func (rs *Records) RecordOf(instance, milestone string) ([]recordView, error) { + if instance != "" { + r, err := rs.Read(instance) + if err != nil { + return nil, err + } + if r == nil { + return nil, fmt.Errorf("no delivery record %s is kept", instance) + } + return []recordView{viewOf(r)}, nil + } + if milestone == "" { + return nil, errors.New("name a record's instance or a milestone") + } + list, err := rs.List() + if err != nil { + return nil, err + } + var out []recordView + for _, r := range list { + if r.Milestone == milestone { + out = append(out, viewOf(r)) + } + } + if len(out) == 0 { + return nil, fmt.Errorf("no delivery record of the milestone %s is kept", milestone) + } + return out, nil +} + +// recordLine is one record as gate_records lists it. +type recordLine struct { + Instance string `json:"instance"` + Milestone string `json:"milestone"` + Opened string `json:"opened"` + Ended string `json:"ended,omitempty"` + ForgeMade bool `json:"forge-made,omitempty"` + Entries int `json:"entries"` + Outcomes int `json:"outcomes"` + Away int `json:"away,omitempty"` + Pinned int `json:"pinned,omitempty"` + Bytes int `json:"bytes"` + Of int `json:"of-bytes"` + Waits string `json:"waits,omitempty"` +} + +// Lines are every record kept, oldest first, each with its fill against the value limit. +func (rs *Records) Lines() (map[string]any, error) { + list, err := rs.List() + if err != nil { + return nil, err + } + lines := make([]recordLine, 0, len(list)) + open := 0 + for _, r := range list { + v := viewOf(r) + l := recordLine{Instance: r.Instance, Milestone: r.Milestone, Opened: timeText(&r.Opened), ForgeMade: r.ForgeMade, + Entries: len(r.Entries), Outcomes: len(r.Outcomes), Away: len(r.Away), Pinned: len(r.Pinned), Bytes: v.Bytes, + Of: v.Of, Waits: v.Waits} + if r.Ended != nil { + l.Ended = timeText(&r.Ended.At) + ": " + r.Ended.Why + } else { + open++ + } + lines = append(lines, l) + } + return map[string]any{"records": lines, "open": open, "kept": len(lines), "retention-days": int(rs.retention().Hours() / 24)}, nil +} + +// EndByHand ends one open record by hand, with the reason, recorded in the hand-act log first: nothing is ended when the +// log did not take it. +func (rs *Records) EndByHand(instance, why, by string, acts handActs) (recordView, error) { + if strings.TrimSpace(why) == "" { + return recordView{}, errors.New("say why the record is ended by hand, such as a node gone for good") + } + r, err := rs.Read(instance) + if err != nil { + return recordView{}, err + } + if r == nil { + return recordView{}, fmt.Errorf("no delivery record %s is kept", instance) + } + if r.Ended != nil { + return viewOf(r), fmt.Errorf("the record %s ended before, at %s: %s", instance, timeText(&r.Ended.At), r.Ended.Why) + } + if acts == nil { + return recordView{}, errors.New("the hand-act log cannot be reached here, so nothing was ended") + } + if _, err := acts.Record(meshhandactsclient.Request{What: "ended the delivery record " + instance + " by hand", + Why: why, Cause: "gate_record_end", Verb: "record-end"}); err != nil { + return recordView{}, fmt.Errorf("the hand act was not recorded through mesh-hand-acts, so nothing was ended: %w", err) + } + if err := rs.End(r, why, by); err != nil { + return recordView{}, err + } + rs.keep(r) + return viewOf(r), nil +} + +// handActs records an act in the hand-act log (meshhandactsclient.Recorder). +type handActs interface { + Record(q meshhandactsclient.Request) (meshhandactsclient.Taken, error) +} + +// recordTools are the record's tools; acts is the hand-act log. +func recordTools(h *Holder, acts handActs) []stdio.Tool { + none := func() (any, error) { return nil, errors.New("this gate keeps no delivery records") } + return []stdio.Tool{ + {Name: "gate_record", + Description: "One delivery record whole (design 47, the delivery record): its group's state, its trace, each " + + "entry's outcome per node, the nodes away with since when, the revisions it pins, how it ended, its size " + + "against the bus's value limit, and while open what it waits for. By its instance, or every record of a milestone.", + Input: map[string]any{"type": "object", "properties": map[string]any{ + "instance": str("a record's instance, record//"), "milestone": str("a milestone")}}, + Run: func(a map[string]any) (any, error) { + if h.Records == nil { + return none() + } + return h.Records.RecordOf(strArg(a, "instance"), strArg(a, "milestone")) + }}, + {Name: "gate_records", + Description: "Every delivery record kept, oldest first: its milestone, when it opened and ended, its entries, " + + "outcomes, nodes away and pins, its size against the bus's value limit, and while open what it waits for.", + Run: func(map[string]any) (any, error) { + if h.Records == nil { + return none() + } + return h.Records.Lines() + }}, + {Name: "gate_consumer", + Description: "The delivery record's consumer of the step events: per family the events applied and refused " + + "since the gate started, the last applied, and why the last was refused.", + Run: func(map[string]any) (any, error) { + if h.Consumer == nil { + return nil, errors.New("this gate applies no step events") + } + return map[string]any{"families": h.Consumer.Heard()}, nil + }}, + {Name: "gate_record_end", + Description: "End a delivery record left open by hand (a node gone for good, an entry that never runs), with the " + + "reason; recorded in the hand-act log of the seat mesh-hand-acts first, and said as record-ended.", + Input: map[string]any{"type": "object", "properties": map[string]any{ + "instance": str("the record's instance"), "why": str("why it is ended by hand, in plain words")}, + "required": []string{"instance", "why"}}, + Run: func(a map[string]any) (any, error) { + if h.Records == nil { + return none() + } + return h.Records.EndByHand(strArg(a, "instance"), strArg(a, "why"), stdio.CallerOf(a), acts) + }}, + } +} diff --git a/modules/mesh-delivery-gate/go.mod b/modules/mesh-delivery-gate/go.mod index e352605ad..2b9665c4d 100644 --- a/modules/mesh-delivery-gate/go.mod +++ b/modules/mesh-delivery-gate/go.mod @@ -3,6 +3,7 @@ module mesh-delivery-gate go 1.26.0 require ( + git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliverycheckerclient v0.0.0-00010101000000-000000000000 @@ -32,6 +33,7 @@ require ( ) require ( + git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshhandacts v0.0.0-00010101000000-000000000000 // indirect git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient v0.0.0-00010101000000-000000000000 // indirect git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient v0.0.0-00010101000000-000000000000 // indirect git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner v0.0.0-00010101000000-000000000000 // indirect diff --git a/modules/mesh-delivery-gate/module.json b/modules/mesh-delivery-gate/module.json index a619a7d8e..b10a9275a 100644 --- a/modules/mesh-delivery-gate/module.json +++ b/modules/mesh-delivery-gate/module.json @@ -257,7 +257,8 @@ "seat:git.statuses", "seat:git.open-pulls", "seat:git.contains", - "gitea.gitea_commit_statuses" + "gitea.gitea_commit_statuses", + "seat:mesh-hand-acts.hand-act" ], "state": [ "deliveries", @@ -294,7 +295,11 @@ "delivery_hand_now", "delivery_merge_again", "delivery_status", - "delivery_turns" + "delivery_turns", + "gate_consumer", + "gate_record", + "gate_record_end", + "gate_records" ], "config": { "refusal-stall-floor-minutes": {