diff --git a/checks/health/go.mod b/checks/health/go.mod index 5ec7f46c..b5ae8753 100644 --- a/checks/health/go.mod +++ b/checks/health/go.mod @@ -2,4 +2,13 @@ module checks/health go 1.26.0 -require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030001116-2da0de75c3f0 +require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031081231-d352fe51b455 + +require ( + github.com/klauspost/compress v1.20.0 // indirect + github.com/nats-io/nats.go v1.54.0 // indirect + 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 + golang.org/x/sys v0.48.0 // indirect +) diff --git a/checks/health/go.sum b/checks/health/go.sum index f8bfbd2b..5668ee75 100644 --- a/checks/health/go.sum +++ b/checks/health/go.sum @@ -1,2 +1,16 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030001116-2da0de75c3f0 h1:lKU3HjNXGiVbLW2hUgFjU/TMTUYMlNnnmZxG5T7biQA= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030001116-2da0de75c3f0/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031081231-d352fe51b455 h1:1rAqVPyu07xkacqJrDaqUtoqLQJmklfyXC/5erRJnEk= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031081231-d352fe51b455/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA= +github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= +github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA= +github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k= +github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg= +github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs= +github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= +github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= +golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= +golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= diff --git a/checks/manifests/go.mod b/checks/manifests/go.mod index 6fbb4f39..82d2cad9 100644 --- a/checks/manifests/go.mod +++ b/checks/manifests/go.mod @@ -2,4 +2,13 @@ module checks/manifests go 1.26.0 -require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030181247-4c1384faefa0 +require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031081231-d352fe51b455 + +require ( + github.com/klauspost/compress v1.20.0 // indirect + github.com/nats-io/nats.go v1.54.0 // indirect + 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 + golang.org/x/sys v0.48.0 // indirect +) diff --git a/checks/manifests/go.sum b/checks/manifests/go.sum index fc5109a2..9cb91cd0 100644 --- a/checks/manifests/go.sum +++ b/checks/manifests/go.sum @@ -1,2 +1,16 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030181247-4c1384faefa0 h1:t04baH2Xz/NUsOEV3ctKY59XjiY5WmEuststvH+XeOs= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030181247-4c1384faefa0/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031081231-d352fe51b455 h1:1rAqVPyu07xkacqJrDaqUtoqLQJmklfyXC/5erRJnEk= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031081231-d352fe51b455/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA= +github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= +github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA= +github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k= +github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg= +github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs= +github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= +github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= +golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= +golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= diff --git a/modules/mesh-hand-acts/README.md b/modules/mesh-hand-acts/README.md index b435e08d..3a13fdb4 100644 --- a/modules/mesh-hand-acts/README.md +++ b/modules/mesh-hand-acts/README.md @@ -22,5 +22,12 @@ alone writes the log. - `mesh-hand-acts.hand-acts`: both logs, this one and the controller's `mesh-controller_hand-acts` until its lease ends, newest first, with the causes repaired by hand twice within a fortnight; `warrants=true` lists the warrants. +- `hand_acts_copied` and `hand_acts_copy` (research 072 row 26): the controller's hand acts are copied once into this + seat's log, each with its own key, time and author. It runs at the module's start while the old bucket is readable + and no marker (`copied`, key `controller-hand-acts`) says it was done, and by the tool; an act already in the log is + never appended again, and `hand-acts` counts a copied act once. `hand_acts_copied` says whether, when and how many. + The lease write has no hook to run it from, so the start is the trigger; `hand_acts_copy` with `top_up=true`, asked + after the lease write, appends any act the controller wrote since the copy (a finished copy is otherwise a no-op). + No act is written to both logs: every recorder calls this seat from its delivery, and the controller keeps its own acts in its own bucket until it loses its lease. diff --git a/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts.go b/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts.go index 6e8579e8..120a736f 100644 --- a/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts.go +++ b/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts.go @@ -18,6 +18,7 @@ import ( "errors" "fmt" "regexp" + "slices" "sort" "strconv" "strings" @@ -287,8 +288,18 @@ func (l *Log) HandActs(days int, cause, who string) Listing { since := now.Add(-time.Duration(days) * 24 * time.Hour) out := Listing{Acts: []Act{}, Repeated: map[string]int{}, Days: days, Kept: map[string]int{}} var all []Act + // What was copied from the controller's log is in this one: it is read from this one only (copy.go). + copied := map[string]bool{} for _, log := range []string{ActsLog, ControllerActs} { acts, kept, err := l.read(log, now.Add(-max(time.Duration(days)*24*time.Hour, RepeatedWithin))) + if log == ActsLog && err == nil { + for _, a := range acts { + copied[a.ID] = a.ID != "" + } + } + if log == ControllerActs { + acts = slices.DeleteFunc(acts, func(a Act) bool { return a.ID != "" && copied[a.ID] }) + } if err != nil { if out.Unread == nil { out.Unread = map[string]string{} diff --git a/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts_test.go b/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts_test.go index 73d42cef..ea501dad 100644 --- a/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts_test.go +++ b/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts_test.go @@ -3,6 +3,7 @@ package main import ( "encoding/json" "errors" + "fmt" "sort" "strings" "sync" @@ -284,3 +285,87 @@ func TestHandActsReadsBothLogs(t *testing.T) { t.Errorf("warrants answered %v, %v", w, err) } } + +// The controller's hand acts are copied into this seat's log once, each with its own key, time and author, and +// never twice: a marker makes it once, a key already in the log is not appended again, a log that cannot be read or +// written leaves no marker so the next call copies again, and hand-acts counts a copied act once (research 072 row 26). +func TestTheControllersHandActsAreCopiedOnce(t *testing.T) { + base := time.Date(2026, 10, 28, 0, 0, 0, 0, time.UTC) + old := []string{fmt.Sprintf("act-%d-1", base.UnixNano()), fmt.Sprintf("act-%d-2", base.UnixNano()+1), + fmt.Sprintf("act-%d-3", base.UnixNano()+2), "warrant-x-y"} + cases := []struct { + name string + already []string // keys in the new log before the copy + oldErr, newErr error + junk bool + wantErr string + wantCopied, wantAlready int + wantMarker bool + }{ + {"all", nil, nil, nil, false, "", 4, 0, true}, + {"finishes a half copy", []string{old[0], old[1]}, nil, nil, false, "", 2, 2, true}, + {"an entry that is no act is copied unchanged", nil, nil, nil, true, "", 5, 0, true}, + {"the old log cannot be read", nil, errors.New("away"), nil, false, "cannot be read", 0, 0, false}, + {"the new log cannot be written", nil, nil, errors.New("full"), false, "could not be copied", 0, 0, false}, + } + for _, tc := range cases { + l, store, _, _ := newLog() + for _, k := range old { + _ = store.Put(ControllerActs, k, Act{ID: k, At: base, By: "ace.mesh-config", Verb: VerbRecord, Why: "w", Cause: "c"}) + } + if tc.junk { + _ = store.Put(ControllerActs, fmt.Sprintf("act-%d-4", base.UnixNano()+3), "not an act") + } + for _, k := range tc.already { + _ = store.Append(ActsLog, k, Act{ID: k}) + } + store.keysErr[ControllerActs], store.keysErr[ActsLog] = tc.oldErr, tc.newErr + st, err := l.Copy("test", false) + if tc.wantErr != "" { + if err == nil || !strings.Contains(err.Error(), tc.wantErr) { + t.Errorf("%s: %v", tc.name, err) + } + } else if err != nil || !st.Done || st.Marker.Copied != tc.wantCopied || st.Marker.Already != tc.wantAlready { + t.Fatalf("%s: %+v, %v", tc.name, st, err) + } + _, marked, _ := store.Get(CopiedState, CopyKey) + if marked != tc.wantMarker || len(store.logKeys(ActsLog)) != map[bool]int{true: tc.wantCopied + tc.wantAlready, false: len(tc.already)}[tc.wantMarker] { + t.Errorf("%s: marker %v, log %v", tc.name, marked, store.logKeys(ActsLog)) + } + if !tc.wantMarker { + continue + } + // Each act is kept with its own time and author, as the controller wrote it. + raw, _, _ := store.Last(ActsLog, old[2]) + var kept Act + _ = json.Unmarshal(raw, &kept) + if kept.By != "ace.mesh-config" || !kept.At.Equal(base) { + t.Errorf("%s: kept %+v", tc.name, kept) + } + // Asked again: nothing more is appended, and hand-acts counts each act once. + before := len(store.logKeys(ActsLog)) + if again, err := l.Copy("test", false); err != nil || !again.Done || !strings.Contains(again.Said, "already copied") || + len(store.logKeys(ActsLog)) != before { + t.Errorf("%s: a second copy: %+v, %v", tc.name, again, err) + } + if got := l.HandActs(30, "", ""); got.Kept[ControllerActs] != len(old)+boolInt(tc.junk) || len(got.Acts) != len(old) { + t.Errorf("%s: %d acts listed from %d kept: %+v", tc.name, len(got.Acts), before, got.Kept) + } + // An act the controller wrote since is appended by a top-up, and only that one. + late := fmt.Sprintf("act-%d-9", base.UnixNano()+9) + _ = store.Put(ControllerActs, late, Act{ID: late, At: base, By: "ace.late", Verb: VerbRecord, Why: "w", Cause: "c"}) + if up, err := l.Copy("test", true); err != nil || up.Marker.Copied != 1 || len(store.logKeys(ActsLog)) != before+1 { + t.Errorf("%s: a top-up: %+v, %v", tc.name, up, err) + } + if s, err := l.Copied(); err != nil || !s.Done || s.Marker == nil { + t.Errorf("%s: copied says %+v, %v", tc.name, s, err) + } + } +} + +func boolInt(b bool) int { + if b { + return 1 + } + return 0 +} diff --git a/modules/mesh-hand-acts/cmd/mesh-hand-acts/copy.go b/modules/mesh-hand-acts/cmd/mesh-hand-acts/copy.go new file mode 100644 index 00000000..0c289388 --- /dev/null +++ b/modules/mesh-hand-acts/cmd/mesh-hand-acts/copy.go @@ -0,0 +1,183 @@ +package main + +// The one-time copy of the controller's hand-act log into this seat's log (novox/hq research 072 row 26, ADR 0571 +// §13). The controller kept every act of its life in its bucket `mesh-controller_hand-acts` (574 on 2026-10-31); from +// the lease write only this seat reads them, so they are copied once into `acts`, each as the controller wrote it: the +// same key, the same bytes, so its time and its author are what they were. +// +// **When it runs.** The lease write has no hook to run from (it is a person's step, ADR 0561 §6), so the copy runs at +// this module's start while the old log is readable and no marker says it was copied, and on the tool +// `hand_acts_copy`. Both do the same thing, and neither does it twice (a `top_up` of the tool appends only what is new): the marker `controller-hand-acts` of the state +// `copied` is created once, after every act is in the log; an act already in the log (by its key) is never appended +// again, so a copy that stopped half way is finished by the next one. Until the marker exists `hand-acts` hides an +// old act that is already copied, so the two logs never count one act twice. `hand_acts_copied` says where it stands. + +import ( + "encoding/json" + "errors" + "fmt" + "sort" + "time" +) + +// CopiedState is the state that holds the marker, and CopyKey the marker's key. +const ( + CopiedState = "copied" + CopyKey = "controller-hand-acts" +) + +// CopyMarker is what the copy leaves behind once it is whole: when, by what, how many acts, and which entries could not +// be read as acts (copied unchanged all the same, so nothing is lost). +type CopyMarker struct { + At time.Time `json:"at"` + By string `json:"by"` + Source string `json:"source"` + Found int `json:"found"` + Copied int `json:"copied"` + Already int `json:"already-there"` + // Unreadable are the keys whose value is not an act; they were copied as they were. + Unreadable []string `json:"unreadable,omitempty"` + FirstKey string `json:"first-key,omitempty"` + LastKey string `json:"last-key,omitempty"` +} + +// CopyStatus is what the tools answer: the marker if the copy is whole, and how the two logs stand now. +type CopyStatus struct { + Done bool `json:"done"` + Marker *CopyMarker `json:"marker,omitempty"` + // Said is a sentence for a person: what this call did or found. + Said string `json:"said"` + // Pending are the old acts not yet in this seat's log (counted only while the copy is not whole). + Pending *int `json:"pending,omitempty"` + Old *int `json:"old-log-keys,omitempty"` + New *int `json:"new-log-entries,omitempty"` +} + +// Copy copies the old log into this one, once. by says what asked (`start` or a caller). With topUp it also runs after +// the marker exists, appending only the acts the log does not hold yet (the controller's, written since the copy), and +// leaves the marker as it was. An old log that cannot be +// read, or an act that cannot be appended, is an error and leaves no marker, so the next call copies again. +func (l *Log) Copy(by string, topUp bool) (CopyStatus, error) { + marked := false + if raw, found, err := l.Store.Get(CopiedState, CopyKey); err != nil { + return CopyStatus{}, fmt.Errorf("whether the controller's hand acts were copied cannot be read: %w", err) + } else if found { + var m CopyMarker + if err := json.Unmarshal(raw, &m); err != nil { + return CopyStatus{}, fmt.Errorf("the copy's marker is not readable: %w", err) + } + if !topUp { + return CopyStatus{Done: true, Marker: &m, Said: fmt.Sprintf("already copied at %s: %d acts of %s, nothing done", + m.At.Format(time.RFC3339), m.Copied+m.Already, m.Source)}, nil + } + marked = true + } + keys, err := l.Store.Keys(ControllerActs) + if err != nil { + return CopyStatus{}, fmt.Errorf("the controller's hand acts cannot be read, so none were copied: %w", err) + } + sortByTime(keys) + m := CopyMarker{By: by, Source: ControllerActs, Found: len(keys)} + l.Logf("[%s] copying the controller's hand acts into %s: %d found in %s", Module, ActsLog, len(keys), ControllerActs) + for i, k := range keys { + raw, found, err := l.Store.Get(ControllerActs, k) + if err != nil { + return CopyStatus{}, fmt.Errorf("%s of %s cannot be read after %d of %d acts were copied: %w", k, ControllerActs, m.Copied, len(keys), err) + } + if !found { + continue // removed since it was listed + } + if _, there, err := l.Store.Last(ActsLog, k); err != nil { + return CopyStatus{}, fmt.Errorf("whether %s is in %s cannot be read, after %d acts were copied: %w", k, ActsLog, m.Copied, err) + } else if there { + m.Already++ + continue + } + var a Act + if json.Unmarshal(raw, &a) != nil { + m.Unreadable = append(m.Unreadable, k) + l.Logf("[%s] %s of %s is not an act; copied unchanged", Module, k, ControllerActs) + } + if err := l.Store.Append(ActsLog, k, json.RawMessage(raw)); err != nil { + return CopyStatus{}, fmt.Errorf("%s could not be copied into %s after %d acts: %w", k, ActsLog, m.Copied, err) + } + m.Copied++ + if m.FirstKey == "" { + m.FirstKey = k + } + m.LastKey = k + if (i+1)%100 == 0 { + l.Logf("[%s] copied %d of %d", Module, i+1, len(keys)) + } + } + m.At = l.Now().UTC() + if marked { + l.Logf("[%s] topped up the copy: %d acts copied, %d already there, of %d found", Module, m.Copied, m.Already, m.Found) + return CopyStatus{Done: true, Marker: &m, Said: fmt.Sprintf("topped up: %d acts copied that the first copy did not hold (%d already there)", m.Copied, m.Already)}, nil + } + if err := l.Store.Create(CopiedState, CopyKey, m); errors.Is(err, ErrExists) { + return CopyStatus{Said: "another instance finished the copy first; nothing more done"}, nil + } else if err != nil { + return CopyStatus{}, fmt.Errorf("every act was copied but the marker could not be written, so the next call checks them again: %w", err) + } + l.Logf("[%s] copied the controller's hand acts: %d copied, %d already there, %d not acts, of %d found (%s .. %s)", + Module, m.Copied, m.Already, len(m.Unreadable), m.Found, m.FirstKey, m.LastKey) + return CopyStatus{Done: true, Marker: &m, Said: fmt.Sprintf("copied %d acts (%d were already there) of %d found in %s", + m.Copied, m.Already, m.Found, ControllerActs)}, nil +} + +// Copied says where the copy stands without doing anything: the marker, or how many old acts wait. +func (l *Log) Copied() (CopyStatus, error) { + raw, found, err := l.Store.Get(CopiedState, CopyKey) + if err != nil { + return CopyStatus{}, fmt.Errorf("the copy's marker cannot be read: %w", err) + } + var st CopyStatus + if found { + var m CopyMarker + if err := json.Unmarshal(raw, &m); err != nil { + return CopyStatus{}, fmt.Errorf("the copy's marker is not readable: %w", err) + } + st = CopyStatus{Done: true, Marker: &m, Said: fmt.Sprintf("copied at %s by %s: %d copied, %d already there, of %d in %s", + m.At.Format(time.RFC3339), m.By, m.Copied, m.Already, m.Found, m.Source)} + } + if f, err := l.Store.Fill(ActsLog); err == nil { + n := int(f.Entries) + st.New = &n + } + keys, err := l.Store.Keys(ControllerActs) + if err != nil { + if !st.Done { + st.Said = "not copied yet, and the controller's log cannot be read: " + err.Error() + } + return st, nil + } + n := len(keys) + st.Old = &n + if !st.Done { + pending := 0 + for _, k := range keys { + if _, there, err := l.Store.Last(ActsLog, k); err != nil || !there { + pending++ + } + } + st.Pending = &pending + st.Said = fmt.Sprintf("not copied yet: %d of %d old acts wait; hand_acts_copy copies them", pending, n) + } + return st, nil +} + +// sortByTime orders keys oldest first by the time they name, those that name none after, by key. +func sortByTime(keys []string) { + sort.SliceStable(keys, func(i, j int) bool { + ti, oi := keyTime(keys[i]) + tj, oj := keyTime(keys[j]) + switch { + case oi && oj && !ti.Equal(tj): + return ti.Before(tj) + case oi != oj: + return oi + } + return keys[i] < keys[j] + }) +} diff --git a/modules/mesh-hand-acts/cmd/mesh-hand-acts/main.go b/modules/mesh-hand-acts/cmd/mesh-hand-acts/main.go index b103fa23..7c1a07c0 100644 --- a/modules/mesh-hand-acts/cmd/mesh-hand-acts/main.go +++ b/modules/mesh-hand-acts/cmd/mesh-hand-acts/main.go @@ -77,6 +77,15 @@ func (runtimeStore) Fill(log string) (Fill, error) { func main() { l := &Log{Store: runtimeStore{}, Now: time.Now, Logf: logf, Publish: stdio.SeatPublish} + // No hook of the lease write runs this module, so the one-time copy of the controller's hand acts runs at start + // and on the tool hand_acts_copy; the marker makes it once (copy.go). + go func() { + if st, err := l.Copy("start", false); err != nil { + logf("[%s] the copy of the controller's hand acts is NOT done: %v", Module, err) + } else { + logf("[%s] %s", Module, st.Said) + } + }() if err := stdio.Serve("", Tools(l)); err != nil { logf("%v", err) os.Exit(1) diff --git a/modules/mesh-hand-acts/cmd/mesh-hand-acts/manifest_test.go b/modules/mesh-hand-acts/cmd/mesh-hand-acts/manifest_test.go index cd19669a..05c9d53b 100644 --- a/modules/mesh-hand-acts/cmd/mesh-hand-acts/manifest_test.go +++ b/modules/mesh-hand-acts/cmd/mesh-hand-acts/manifest_test.go @@ -47,7 +47,7 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { t.Errorf("emits %v, reads %v", m.Emits, m.Reads) } readBy := map[string]string{ActsLog: HandActsTool, WarrantsState: HandActsTool, ControllerActs: HandActsTool, - RouterRecords: WarrantedTool} + RouterRecords: WarrantedTool, CopiedState: CopiedTool} tools := map[string]bool{} for _, tool := range Tools(&Log{}) { tools[tool.Name] = true @@ -58,8 +58,8 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { } for _, raw := range m.State { var name string - if json.Unmarshal(raw, &name) != nil || name != WarrantsState { - t.Errorf("the state %s is not the warrants, or ages", raw) + if json.Unmarshal(raw, &name) != nil || (name != WarrantsState && name != CopiedState) { + t.Errorf("the state %s is neither the warrants nor the copy's marker, or ages", raw) } if !tools[readBy[name]] { t.Errorf("the state %s has no tool that reads it", name) diff --git a/modules/mesh-hand-acts/cmd/mesh-hand-acts/tools.go b/modules/mesh-hand-acts/cmd/mesh-hand-acts/tools.go index 833db625..ad51f7ed 100644 --- a/modules/mesh-hand-acts/cmd/mesh-hand-acts/tools.go +++ b/modules/mesh-hand-acts/cmd/mesh-hand-acts/tools.go @@ -21,6 +21,9 @@ const ( HandActTool = Seat + ".hand-act" HandActsTool = Seat + ".hand-acts" WarrantedTool = Seat + ".warranted" + // CopiedTool reads where the one-time copy of the controller's hand acts stands, CopyTool does it (copy.go). + CopiedTool = "hand_acts_copied" + CopyTool = "hand_acts_copy" ) func obj(props map[string]any, required ...string) map[string]any { @@ -70,6 +73,21 @@ func Tools(l *Log) []stdio.Tool { Run: func(a map[string]any) (any, error) { return l.Warranted(stdio.CallerOf(a), arg(a, "asker"), arg(a, "ask")) }}, + {Name: CopiedTool, + Description: "Where the one-time copy of the controller's hand acts into this seat's log stands (novox/hq research 072 " + + "row 26): done, with when, by what and how many acts were copied, or how many old acts still wait, and how " + + "full both logs are. Changes nothing.", + Input: obj(map[string]any{}), + Run: func(map[string]any) (any, error) { return l.Copied() }}, + {Name: CopyTool, + Description: "Copy the controller's hand acts (its bucket mesh-controller_hand-acts) into this seat's log once, each " + + "with its own key, time and author. Does nothing when the copy is already done; an act already in the log is " + + "never appended again, so a copy that stopped is finished by asking again. With top_up=true it also runs after the " + + "copy is done and appends only the acts written since. This also runs at the module's start.", + Input: obj(map[string]any{"top_up": str("\"true\": after the copy is done, append the acts written since")}), + Run: func(a map[string]any) (any, error) { + return l.Copy("tool, called by "+stdio.CallerOf(a), a["top_up"] == true || strings.EqualFold(arg(a, "top_up"), "true")) + }}, } } diff --git a/modules/mesh-hand-acts/cmd/mesh-hand-acts/whole_integration_test.go b/modules/mesh-hand-acts/cmd/mesh-hand-acts/whole_integration_test.go index 64ec46d2..8977994d 100644 --- a/modules/mesh-hand-acts/cmd/mesh-hand-acts/whole_integration_test.go +++ b/modules/mesh-hand-acts/cmd/mesh-hand-acts/whole_integration_test.go @@ -338,7 +338,7 @@ func TestTheHandActLogWholeRecordsOnTheBus(t *testing.T) { if err != nil { t.Fatal(err) } - for _, b := range []string{Module + "_" + WarrantsState, "mesh-controller_hand-acts", "messenger_asks"} { + for _, b := range []string{Module + "_" + WarrantsState, Module + "_" + CopiedState, "mesh-controller_hand-acts", "messenger_asks"} { if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b}); err != nil { t.Fatal(err) } @@ -354,6 +354,13 @@ func TestTheHandActLogWholeRecordsOnTheBus(t *testing.T) { if err != nil { t.Fatal(err) } + // One act of the controller's, kept before the module starts: the start copies it into the log. + oldActs, _ := js.KeyValue(ctx, "mesh-controller_hand-acts") + oldKey := fmt.Sprintf("act-%d-1", time.Now().Add(-time.Hour).UnixNano()) + if _, err := oldActs.Put(ctx, oldKey, []byte(`{"id":"`+oldKey+`","at":"`+time.Now().Add(-time.Hour).UTC().Format(time.RFC3339)+ + `","by":"ace.old","verb":"hand-act","why":"w","cause":"c"}`)); err != nil { + t.Fatal(err) + } cmd := exec.CommandContext(ctx, bin) cmd.Env = append(os.Environ(), "MESH_NODE=novox", "MESH_MODULE="+Module) stdin, _ := cmd.StdinPipe() @@ -395,8 +402,19 @@ func TestTheHandActLogWholeRecordsOnTheBus(t *testing.T) { t.Fatal(err) } } - if info, err := logStream.Info(ctx); err != nil || info.State.Msgs != 2 { - t.Errorf("the log keeps %v, want the act and one warrant (%v)", info, err) + for i := 0; ; i++ { + if c, err := p.call(CopiedTool, "person.jochen", map[string]any{}); err == nil && strings.Contains(string(c), `"done":true`) { + break + } else if i > 50 { + t.Fatalf("the controller's hand acts were not copied at the start: %s, %v", c, err) + } + time.Sleep(200 * time.Millisecond) + } + if e, err := logStream.GetLastMsgForSubject(ctx, "mesh.log."+Module+"."+ActsLog+"."+oldKey); err != nil || !strings.Contains(string(e.Data), `"by":"ace.old"`) { + t.Errorf("the controller's act is not copied with its author: %v", err) + } + if info, err := logStream.Info(ctx); err != nil || info.State.Msgs != 3 { + t.Errorf("the log keeps %v, want the copied act, the act and one warrant (%v)", info, err) } listed, err := p.call(HandActsTool, "person.jochen", map[string]any{}) if err != nil || !strings.Contains(string(listed), "warrant-claude-code-a1") || !strings.Contains(string(listed), r.Act.ID) { diff --git a/modules/mesh-hand-acts/module.json b/modules/mesh-hand-acts/module.json index 96aff8b1..443ea9d3 100644 --- a/modules/mesh-hand-acts/module.json +++ b/modules/mesh-hand-acts/module.json @@ -24,7 +24,8 @@ } ], "state": [ - "warrants" + "warrants", + "copied" ], "own-secrets": { "broker": "${dir:state}/broker" diff --git a/modules/nats/cmd/nats-tools/manifest_test.go b/modules/nats/cmd/nats-tools/manifest_test.go index 2007b1ef..dc9f453a 100644 --- a/modules/nats/cmd/nats-tools/manifest_test.go +++ b/modules/nats/cmd/nats-tools/manifest_test.go @@ -112,12 +112,12 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) { } // The tools that change the bus (issue 1952 N1-N3): listed, their audit events declared, and the credential // they dial with the module's own secret `broker`. - for _, want := range []string{"nats_consumer_remove", "nats_bucket_remove"} { + for _, want := range []string{"nats_consumer_remove", "nats_bucket_remove", "nats_stream_remove"} { if !slices.Contains(m.Tools, want) { t.Errorf("tools do not list %s", want) } } - for _, want := range []string{"consumer-removed", "bucket-keys-removed"} { + for _, want := range []string{"consumer-removed", "bucket-keys-removed", "stream-removed"} { if !slices.Contains(m.Emits, want) { t.Errorf("emits do not list %s: the removal tools say it as the audit line", want) } diff --git a/modules/nats/cmd/nats-tools/removal.go b/modules/nats/cmd/nats-tools/removal.go index 3aff66e5..b8efc4f2 100644 --- a/modules/nats/cmd/nats-tools/removal.go +++ b/modules/nats/cmd/nats-tools/removal.go @@ -1,7 +1,7 @@ package main -// The two tools of this module that change the bus (novox/hq ADR 0564 clause 5, issue 1952 notes 6, 7 and 9): -// remove one named consumer, and remove one node's keys under a prefix of a key-value bucket. Everything else +// The three tools of this module that change the bus (novox/hq ADR 0564 clause 5, issue 1952 notes 6, 7 and 9): +// remove one named consumer, remove one named stream, and remove one node's keys under a prefix of a key-value bucket. Everything else // here reads. // // **Narrow by what the tool needs, and narrower by what the bus grants.** The tools reach the bus on the @@ -34,6 +34,7 @@ type Said func(event string, body any) error const ( consumerRemoved = "consumer-removed" bucketKeysRemoved = "bucket-keys-removed" + streamRemoved = "stream-removed" // maxKeysShown bounds the keys an answer lists; the count is always whole. maxKeysShown = 200 ) @@ -227,7 +228,85 @@ func RemoveKeys(ctx context.Context, bus Bus, said Said, a map[string]any) (any, return out, nil } -// removalTools are the module's two tools that change the bus; every other tool here reads. +// StreamRemoved is what nats_stream_remove answers and says. +type StreamRemoved struct { + Stream string `json:"stream"` + Why string `json:"why"` + DryRun bool `json:"dry_run"` + Present bool `json:"present"` + Messages uint64 `json:"messages"` + Bytes uint64 `json:"bytes"` + Consumers int `json:"consumers"` + Removed bool `json:"removed"` + Said string `json:"said"` + At time.Time `json:"at"` +} + +// RemoveStream removes one named stream whole, after reading it, and says so (the lease write's step for the stream +// ASSIGNMENTS, novox/hq research 072 row 33). **A stream already gone is no error**: the removal is the same act +// however often it is asked, and says it found nothing. A stream with a consumer is refused, naming how many: a reader +// is still attached, and that is for a person to look at before the stream goes. +func RemoveStream(ctx context.Context, bus Bus, said Said, a map[string]any) (any, error) { + w, err := why(a) + if err != nil { + return nil, err + } + stream := arg(a, "stream") + if !name.MatchString(stream) { + return nil, errors.New("stream is required, one stream's name (letters, digits, - and _): nothing was removed") + } + dry, _ := a["dry_run"].(bool) + js, closeBus, err := bus(ctx) + if err != nil { + return nil, err + } + defer closeBus() + out := StreamRemoved{Stream: stream, Why: w, DryRun: dry, At: time.Now().UTC()} + sctx, cancel := step(ctx) + s, err := js.Stream(sctx, stream) + cancel() + if errors.Is(err, jetstream.ErrStreamNotFound) { + out.Said = stream + " is not there: nothing to remove" + return out, nil + } + if err != nil { + return nil, unanswered(err, fmt.Sprintf("the stream %s could not be read, so it was not removed", stream)) + } + sctx, cancel = step(ctx) + info, err := s.Info(sctx) + cancel() + if err != nil { + return nil, unanswered(err, fmt.Sprintf("the stream %s could not be read, so it was not removed", stream)) + } + out.Present, out.Messages, out.Bytes, out.Consumers = true, info.State.Msgs, info.State.Bytes, info.State.Consumers + if info.State.Consumers > 0 { + return nil, fmt.Errorf("%s has %d consumers: a reader is attached, so it was not removed", stream, info.State.Consumers) + } + if dry { + out.Said = fmt.Sprintf("%s holds %d messages (%d bytes) and no consumer; it would be removed", stream, out.Messages, out.Bytes) + return out, nil + } + sctx, cancel = step(ctx) + err = js.DeleteStream(sctx, stream) + cancel() + if err != nil { + return nil, unanswered(err, fmt.Sprintf("stream %s was not removed", stream)) + } + sctx, cancel = step(ctx) + _, err = js.Stream(sctx, stream) + cancel() + if err == nil { + return nil, fmt.Errorf("stream %s is still there after the removal was accepted", stream) + } + out.Removed = true + out.Said = fmt.Sprintf("%s removed: %d messages (%d bytes) and no consumer", stream, out.Messages, out.Bytes) + if err := said(streamRemoved, out); err != nil { + return nil, fmt.Errorf("stream %s WAS removed (%s) but the audit event could not be said: %w", stream, w, err) + } + return out, nil +} + +// removalTools are the module's three tools that change the bus; every other tool here reads. func removalTools(bus Bus, said Said) []stdio.Tool { run := func(f func(context.Context, Bus, Said, map[string]any) (any, error)) func(map[string]any) (any, error) { return func(a map[string]any) (any, error) { return f(context.Background(), bus, said, a) } @@ -237,6 +316,10 @@ func removalTools(bus Bus, said Said) []stdio.Tool { Description: "Removes (deletes) one named JetStream consumer of one stream, e.g. the consumer `controller` of DEAD_LETTER_NOTICES at the lease write (novox/hq ADR 0564 clause 5). Needs `why`, kept in the audit trail; refuses an unknown stream or consumer; reads the consumer first and answers what it was (filters, pending, unacknowledged); the stream and its other consumers are not touched. The bus refuses any consumer the module's credential is not granted.", Input: map[string]any{"stream": str("the stream, e.g. DEAD_LETTER_NOTICES"), "consumer": str("the consumer's name, e.g. controller"), "why": str("why it is removed, for the audit trail (required)")}, Run: run(RemoveConsumer)}, + {Name: "nats_stream_remove", + Description: "Removes (deletes) one named JetStream stream whole, e.g. ASSIGNMENTS at the lease write, beside the consumer of DEAD_LETTER_NOTICES (novox/hq research 072 row 33). Needs `why`, kept in the audit trail; reads the stream first and answers its messages, bytes and consumers; refuses a stream that still has a consumer; a stream already gone is answered as such and is no error, so asking again is safe. `dry_run` only reads. The bus refuses any stream the module's credential is not granted.", + Input: map[string]any{"stream": str("the stream, e.g. ASSIGNMENTS"), "why": str("why it is removed, for the audit trail (required)"), "dry_run": map[string]any{"type": "boolean", "description": "read the stream, remove nothing"}}, + Run: run(RemoveStream)}, {Name: "nats_bucket_remove", Description: "Removes (purges) keys from a key-value bucket: one node's keys (`node`, the keys starting `.`) or the keys under a `prefix`, e.g. a full acts bucket of the node-engine. Needs `why`, kept in the audit trail; never the whole bucket. Counts the keys first and lists them (`dry_run` only counts; `expect` refuses when the count differs); removes only those keys, and their history. See the buckets and their keys with nats_buckets. The bus refuses any bucket the module's credential is not granted.", Input: map[string]any{"bucket": str("the bucket as nats_buckets lists it, e.g. mesh-host_acts"), "node": str("a node whose keys (`.`) go"), "prefix": str("key prefix instead of node"), diff --git a/modules/nats/cmd/nats-tools/removal_test.go b/modules/nats/cmd/nats-tools/removal_test.go index 2ed7799d..eb81522a 100644 --- a/modules/nats/cmd/nats-tools/removal_test.go +++ b/modules/nats/cmd/nats-tools/removal_test.go @@ -31,6 +31,8 @@ accounts { MESH { jetstream: enabled, users: [ publish: { allow: [ "$JS.API.CONSUMER.INFO.DEAD_LETTER_NOTICES.controller", "$JS.API.CONSUMER.DELETE.DEAD_LETTER_NOTICES.controller", + "$JS.API.STREAM.INFO.ASSIGNMENTS", + "$JS.API.STREAM.DELETE.ASSIGNMENTS", "$JS.API.STREAM.INFO.KV_mesh-host_acts", "$KV.mesh-host_acts.>" ] }, subscribe: { allow: ["_INBOX.nats-tools.>"] } } } @@ -73,6 +75,14 @@ accounts { MESH { jetstream: enabled, users: [ if err != nil { t.Fatal(err) } + if _, err := ajs.CreateStream(ctx, jetstream.StreamConfig{Name: "ASSIGNMENTS", Subjects: []string{"mesh.assignment.*.*"}, MaxMsgsPerSubject: 1}); err != nil { + t.Fatal(err) + } + for _, sub := range []string{"mesh.assignment.g14.a", "mesh.assignment.ace.b"} { + if _, err := ajs.Publish(ctx, sub, []byte("{}")); err != nil { + t.Fatal(err) + } + } other, _ := ajs.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "other_state"}) for _, k := range []string{"g14.1", "g14.2", "g14.sub.3", "g14x.4", "ace.5"} { acts.Put(ctx, k, []byte("v")) @@ -131,6 +141,30 @@ accounts { MESH { jetstream: enabled, users: [ t.Fatalf("a consumer already gone: %v", err) } + // Research 072 row 33: the stream ASSIGNMENTS goes whole, once; read first, and asking again is no error; a stream + // outside the grant is not touched. + if _, err := RemoveStream(ctx, bus, say, map[string]any{"stream": "ASSIGNMENTS"}); err == nil || !strings.Contains(err.Error(), "why is required") { + t.Fatalf("a stream without why: %v", err) + } + if _, err := RemoveStream(ctx, bus, say, map[string]any{"stream": "DEAD_LETTER_NOTICES", "why": "x"}); err == nil || !strings.Contains(err.Error(), "grant") { + t.Fatalf("a stream outside the grant: %v", err) + } + if d, err := RemoveStream(ctx, bus, say, map[string]any{"stream": "ASSIGNMENTS", "why": "look", "dry_run": true}); err != nil || d.(StreamRemoved).Messages != 2 || d.(StreamRemoved).Removed { + t.Fatalf("a dry run: %+v %v", d, err) + } + if _, err := ajs.Stream(ctx, "ASSIGNMENTS"); err != nil { + t.Fatalf("a dry run removed the stream: %v", err) + } + if r, err := RemoveStream(ctx, bus, say, map[string]any{"stream": "ASSIGNMENTS", "why": "the lease write"}); err != nil || !r.(StreamRemoved).Removed || r.(StreamRemoved).Messages != 2 { + t.Fatalf("the removal: %+v %v", r, err) + } + if _, err := ajs.Stream(ctx, "ASSIGNMENTS"); err == nil { + t.Fatal("ASSIGNMENTS is still there") + } + if r, err := RemoveStream(ctx, bus, say, map[string]any{"stream": "ASSIGNMENTS", "why": "again"}); err != nil || r.(StreamRemoved).Removed || r.(StreamRemoved).Present { + t.Fatalf("a second removal: %+v %v", r, err) + } + // N3: only the node's keys go; counted first; refused without why, for a whole bucket, and outside the grant. if _, err := RemoveKeys(ctx, bus, say, map[string]any{"bucket": "mesh-host_acts", "node": "g14"}); err == nil || !strings.Contains(err.Error(), "why is required") { t.Fatalf("without why: %v", err) @@ -167,7 +201,7 @@ accounts { MESH { jetstream: enabled, users: [ if k := keys(other); k != "g14.1" { t.Fatalf("another bucket changed: %s", k) } - if strings.Join(said, ",") != "consumer-removed,bucket-keys-removed" { + if strings.Join(said, ",") != "consumer-removed,stream-removed,bucket-keys-removed" { t.Fatalf("the audit events said: %v", said) } } diff --git a/modules/nats/cmd/nats-tools/tools_test.go b/modules/nats/cmd/nats-tools/tools_test.go index 2a2eca4a..7d1a8cdc 100644 --- a/modules/nats/cmd/nats-tools/tools_test.go +++ b/modules/nats/cmd/nats-tools/tools_test.go @@ -234,6 +234,10 @@ func TestTheRemovalToolsRefuseBeforeTheyReachTheBus(t *testing.T) { {"nats_bucket_remove", map[string]any{"bucket": "mesh-host_acts", "node": "g14", "prefix": "g", "why": "x"}, "exactly one of node"}, {"nats_bucket_remove", map[string]any{"bucket": "mesh-host_acts", "prefix": "g*", "why": "x"}, "without wildcards"}, {"nats_bucket_remove", map[string]any{"bucket": "", "node": "g14", "why": "x"}, "bucket is required"}, + {"nats_stream_remove", map[string]any{"stream": "ASSIGNMENTS"}, "why is required"}, + {"nats_stream_remove", map[string]any{"stream": "ASSIGNMENTS", "why": " "}, "why is required"}, + {"nats_stream_remove", map[string]any{"stream": "A.>", "why": "x"}, "stream is required"}, + {"nats_stream_remove", map[string]any{"stream": "", "why": "x"}, "stream is required"}, } { var run func(map[string]any) (any, error) for _, tl := range removalTools(bus, said) { diff --git a/modules/nats/go.mod b/modules/nats/go.mod index 8e261655..cf5a9882 100644 --- a/modules/nats/go.mod +++ b/modules/nats/go.mod @@ -3,7 +3,7 @@ module nats-tools go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030001116-2da0de75c3f0 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031081231-d352fe51b455 github.com/klauspost/compress v1.20.0 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 diff --git a/modules/nats/go.sum b/modules/nats/go.sum index a2b8e018..b669f3e2 100644 --- a/modules/nats/go.sum +++ b/modules/nats/go.sum @@ -1,5 +1,7 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030001116-2da0de75c3f0 h1:lKU3HjNXGiVbLW2hUgFjU/TMTUYMlNnnmZxG5T7biQA= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030001116-2da0de75c3f0/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031081231-d352fe51b455 h1:1rAqVPyu07xkacqJrDaqUtoqLQJmklfyXC/5erRJnEk= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031081231-d352fe51b455/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/nats/module.json b/modules/nats/module.json index 7224731d..81bfc619 100644 --- a/modules/nats/module.json +++ b/modules/nats/module.json @@ -30,7 +30,8 @@ ], "emits": [ "consumer-removed", - "bucket-keys-removed" + "bucket-keys-removed", + "stream-removed" ], "consumes": [], "listens": [ @@ -54,7 +55,8 @@ "nats_user_can", "nats_snapshot_verify", "nats_consumer_remove", - "nats_bucket_remove" + "nats_bucket_remove", + "nats_stream_remove" ], "guards": [ 8222 @@ -208,7 +210,8 @@ ], "bus": { "removes": [ - "DEAD_LETTER_NOTICES.controller" + "DEAD_LETTER_NOTICES.controller", + "ASSIGNMENTS" ] }, "env": {