Merge mesh-catalog main (559939aa) into the registry branch: gate reads forgemsg, messenger narrows asks (K-0; issue 1966)

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
2026-11-05 14:05:04 +01:00
16 changed files with 373 additions and 59 deletions
@@ -7,6 +7,7 @@ import (
"strings"
"sync"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
sdk "git.novox.be/novox/mesh-sdk/go/delivery"
)
@@ -24,7 +25,7 @@ type fakeMerges struct {
manual bool
failed error
// said are the messages the bridge wrote on FORGE, in order.
said []sdk.Forge
said []forgemsg.Forge
}
func newFakeMerges(f *fakeForge) *fakeMerges {
@@ -76,11 +77,11 @@ func (m *fakeMerges) Put(group string, v sdk.Merge) error {
err = f.refuse
}
f.mu.Unlock()
msg := sdk.Forge{Repository: p.Repository, Number: p.Number, Head: p.Head.Head}
msg := forgemsg.Forge{Repository: p.Repository, Number: p.Number, Head: p.Head.Head}
if err != nil {
msg.Kind, msg.Act, msg.Words = sdk.ForgeRefused, "merge", err.Error()
msg.Kind, msg.Act, msg.Words = forgemsg.ForgeRefused, "merge", err.Error()
} else {
msg.Kind, msg.Merge, msg.Root = sdk.ForgeMerged, strings.Repeat("c", 40), "root"
msg.Kind, msg.Merge, msg.Root = forgemsg.ForgeMerged, strings.Repeat("c", 40), "root"
msg.Trees = map[string]string{"app": "tree"}
}
m.say(msg)
@@ -89,12 +90,12 @@ func (m *fakeMerges) Put(group string, v sdk.Merge) error {
}
// say writes a message on FORGE and has the gate hear it, as the runtime does.
func (m *fakeMerges) say(msg sdk.Forge) {
body, err := sdk.EncodeForge(msg)
func (m *fakeMerges) say(msg forgemsg.Forge) {
body, err := forgemsg.EncodeForge(msg)
if err != nil {
panic(err)
}
event, err := sdk.ForgeEvent(msg.Repository, msg.Kind)
event, err := forgemsg.ForgeEvent(msg.Repository, msg.Kind)
if err != nil {
panic(err)
}
@@ -17,6 +17,7 @@ import (
"time"
oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
stdio "git.novox.be/novox/mesh-sdk/go"
sdk "git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/nats-io/nats-server/v2/server"
@@ -585,7 +586,7 @@ func TestAPassingVerdictAsksTheApproveAndItsApproveMergesWithinSecondsOverARealB
t.Fatalf("the gate asked the forge to merge %d times: the bridge merges", n)
}
// The bridge says it merged; the mesh's source then holds the merge, in one write.
b.forgeSays(t, sdk.Forge{Kind: sdk.ForgeMerged, Repository: "novox/app", Number: 7, Head: head, Merge: strings.Repeat("b", 40),
b.forgeSays(t, forgemsg.Forge{Kind: forgemsg.ForgeMerged, Repository: "novox/app", Number: 7, Head: head, Merge: strings.Repeat("b", 40),
Root: "r", Trees: map[string]string{"app": "t"}})
// The forge's own word that the pull request merged follows the bridge's (its webhook): the delivery ends merged.
b.publish(t, "mesh.mod.gitea.event.pull.merged", merged("novox/app", 7, head, strings.Repeat("b", 40)))
@@ -684,13 +685,13 @@ func (b *testBus) answer(t *testing.T, key, option string) time.Time {
}
// forgeSays is the forge bridge's message on FORGE, as its stream carries it.
func (b *testBus) forgeSays(t *testing.T, f sdk.Forge) {
func (b *testBus) forgeSays(t *testing.T, f forgemsg.Forge) {
t.Helper()
raw, err := sdk.EncodeForge(f)
raw, err := forgemsg.EncodeForge(f)
if err != nil {
t.Fatal(err)
}
event, err := sdk.ForgeEvent(f.Repository, f.Kind)
event, err := forgemsg.ForgeEvent(f.Repository, f.Kind)
if err != nil {
t.Fatal(err)
}
@@ -749,7 +750,7 @@ func TestAGroupOfTwoWithGreenChecksGetsOneAskAndOnItsApproveBothMergeOneAfterThe
t.Fatalf("the second member was handed before the first merged: %+v", again.Pulls)
}
one := first.Pulls[0]
b.forgeSays(t, sdk.Forge{Kind: sdk.ForgeMerged, Repository: one.Repository, Number: one.Number, Head: one.Head.Head,
b.forgeSays(t, forgemsg.Forge{Kind: forgemsg.ForgeMerged, Repository: one.Repository, Number: one.Number, Head: one.Head.Head,
Merge: strings.Repeat("a", 40), Root: "r", Trees: map[string]string{"x": "t"}})
var both sdk.Merge
rt.until("the second member to be handed", eventWithin, func() bool { both = mergeAt(b, "feat/x"); return len(both.Pulls) == 2 })
@@ -757,7 +758,7 @@ func TestAGroupOfTwoWithGreenChecksGetsOneAskAndOnItsApproveBothMergeOneAfterThe
t.Fatalf("the second merge handed is %+v", both.Pulls)
}
two := both.Pulls[1]
b.forgeSays(t, sdk.Forge{Kind: sdk.ForgeMerged, Repository: two.Repository, Number: two.Number, Head: two.Head.Head,
b.forgeSays(t, forgemsg.Forge{Kind: forgemsg.ForgeMerged, Repository: two.Repository, Number: two.Number, Head: two.Head.Head,
Merge: strings.Repeat("b", 40), Root: "r", Trees: map[string]string{"x": "t"}})
rt.until("source.main to hold both merges", eventWithin, func() bool {
e, err := b.kvs["source"].Get(ctx, sdk.SourceMainKey)
@@ -6,7 +6,7 @@ import (
"testing"
"time"
sdk "git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
)
// The holder, end to end over fakes (novox/hq ADR 0239 *How it is checked*, cut by ADR 0539 §17): a pull request's head
@@ -241,9 +241,9 @@ func TestAMergedGroupEndsMergedWithSourceMainMoved(t *testing.T) {
t.Fatalf("source.main is %+v, wanted both merges", w.source.main)
}
// A merge that leaves no module.json carries no trees (issue 1367): the gate reads it, and refuses nothing.
msg := sdk.Forge{Kind: sdk.ForgeMerged, Repository: "novox/a", Number: 9, Head: strings.Repeat("e", 40),
msg := forgemsg.Forge{Kind: forgemsg.ForgeMerged, Repository: "novox/a", Number: 9, Head: strings.Repeat("e", 40),
Merge: strings.Repeat("f", 40), Root: "r"}
body, err := sdk.EncodeForge(msg)
body, err := forgemsg.EncodeForge(msg)
if err != nil {
t.Fatalf("a merged message with no trees is refused: %v", err)
}
@@ -251,6 +251,17 @@ func TestAMergedGroupEndsMergedWithSourceMainMoved(t *testing.T) {
if len(w.h.refusals) != 0 {
t.Errorf("the gate refused a merged message with no trees: %v", w.h.refusals)
}
// A group's end is the indexer's: the gate ignores it before decoding, and says no refusal (1585/T25).
w.h.Forged("mesh-delivery-forge-bridge.forge.group.merged", []byte(`{"v":1,"group":"feat/y"}`))
if len(w.h.refusals) != 0 {
t.Errorf("the gate refused a group's end: %v", w.h.refusals)
}
// The repository forge/group's merged message ends the same way: it is decoded as a forge message, and this one,
// which is none, is refused.
w.h.Forged("mesh-delivery-forge-bridge.forge.forge.group.merged", []byte(`{"v":1,"group":"feat/y"}`))
if len(w.h.refusals) != 1 {
t.Errorf("forge/group's merged message was passed by as a group's end: refusals %v", w.h.refusals)
}
}
// ADR 0419 §2, issue 978: a description's `after:` line is not read, and says so once on its pull request. The
@@ -10,6 +10,7 @@ import (
occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient"
oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
sdk "git.novox.be/novox/mesh-sdk/go/delivery"
)
@@ -766,6 +767,6 @@ func allGreenFor(w *world, ms ...member) {
// bridgeMerged is the forge bridge's word that it merged a member.
func bridgeMerged(w *world, repository string, number int, head, merge string) {
w.merges.say(sdk.Forge{Kind: sdk.ForgeMerged, Repository: repository, Number: number, Head: head, Merge: merge, Root: "root",
w.merges.say(forgemsg.Forge{Kind: forgemsg.ForgeMerged, Repository: repository, Number: number, Head: head, Merge: merge, Root: "root",
Trees: map[string]string{"app": "tree"}})
}
@@ -6,6 +6,7 @@ import (
"time"
occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
sdk "git.novox.be/novox/mesh-sdk/go/delivery"
)
@@ -101,7 +102,7 @@ func TestAnAnswerHandsTheFirstMemberOnlyAndTheBridgesMergedHandsTheNext(t *testi
if len(w.merges.puts) != 1 {
t.Fatalf("the next member was handed before the bridge said the first merged: %d merges written", len(w.merges.puts))
}
w.merges.say(sdk.Forge{Kind: sdk.ForgeMerged, Repository: gCat, Number: 2, Head: gHk, Merge: strings.Repeat("a", 40), Root: "r", Trees: map[string]string{"x": "t"}})
w.merges.say(forgemsg.Forge{Kind: forgemsg.ForgeMerged, Repository: gCat, Number: 2, Head: gHk, Merge: strings.Repeat("a", 40), Root: "r", Trees: map[string]string{"x": "t"}})
w.settleAll()
if len(w.merges.puts) != 2 {
t.Fatalf("the second member was not handed after the first merged: %d merges written", len(w.merges.puts))
@@ -116,7 +117,7 @@ func TestAnAnswerHandsTheFirstMemberOnlyAndTheBridgesMergedHandsTheNext(t *testi
// AC2: the bridge's refused hands no member, and the group is said urgent, naming the member and the forge's words.
func TestTheBridgesRefusalHandsNoMemberAndIsSaidUrgent(t *testing.T) {
w := answeredFirst(t)
w.merges.say(sdk.Forge{Kind: sdk.ForgeRefused, Repository: gCat, Number: 2, Head: gHk, Act: "merge",
w.merges.say(forgemsg.Forge{Kind: forgemsg.ForgeRefused, Repository: gCat, Number: 2, Head: gHk, Act: "merge",
Words: "the forge refused gitea_merge_pull_request: mesh/repo-check (missing)"})
w.later(time30s)
w.settleAll()
@@ -138,7 +139,7 @@ func TestTheBridgesRefusalHandsNoMemberAndIsSaidUrgent(t *testing.T) {
// the group, even after the first member merged.
func TestTheNextMemberIsNotHandedWhileARequiredStatusOfTheGroupIsNotASuccess(t *testing.T) {
w := answeredFirst(t)
w.merges.say(sdk.Forge{Kind: sdk.ForgeMerged, Repository: gCat, Number: 2, Head: gHk, Merge: strings.Repeat("a", 40), Root: "r", Trees: map[string]string{"x": "t"}})
w.merges.say(forgemsg.Forge{Kind: forgemsg.ForgeMerged, Repository: gCat, Number: 2, Head: gHk, Merge: strings.Repeat("a", 40), Root: "r", Trees: map[string]string{"x": "t"}})
setChecks(w, gCtl, gHc, "success", "") // the second member's repository check went missing meanwhile
w.settleAll()
if len(w.merges.puts) != 1 {
@@ -194,7 +195,7 @@ func TestARefusalThatRepeatsWithNoTrunkMoveIsNotAskedAgainWithoutEnd(t *testing.
w := answeredFirst(t)
d := w.h.deliveries[IDOf(gCat, gHk)]
refuse := func(took string) {
w.h.ForgeRefused(sdk.Forge{Kind: sdk.ForgeRefused, Repository: gCat, Number: 2, Head: gHk, Act: "merge",
w.h.ForgeRefused(forgemsg.Forge{Kind: forgemsg.ForgeRefused, Repository: gCat, Number: 2, Head: gHk, Act: "merge",
Words: "novox/mesh-catalog#2 is not merged: its verdict names no trunk commit it was judged against, and main is now f70a99dbf485. " +
"The check of head d2d2d2d2d2d2 merged into trunk f70a99dbf485 was asked; the last check took " + took + ". Ask again once both statuses pass"})
}
@@ -10,6 +10,7 @@ import (
"strings"
"time"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
stdio "git.novox.be/novox/mesh-sdk/go"
sdk "git.novox.be/novox/mesh-sdk/go/delivery"
)
@@ -312,23 +313,33 @@ func (h *Holder) mergeValue(name, key string, approved map[string]string, target
return m, len(m.Pulls) > 0 && state == sdk.MergeMerged
}
// forgeGroupEnd is the whole event of a group's end on FORGE, the forge bridge's event name under its module: compared
// exactly, so the merged message of a repository named forge/group, which ends the same way, is read as one.
const forgeGroupEnd = sdk.ForgeBridgeModule + "." + forgemsg.ForgeGroupEndEvent
// Forged takes a message of the forge bridge on FORGE: the gate follows a merge and the refusal of a merge, and nothing else.
//
// The end of a group's merges (`forge.group.merged`, a body this gate has no use for: the gate wrote the merges itself) is
// the indexer's, and is ignored before it is decoded, so no refusal is said for it (issue 1585, ADR 0633 clause 4).
func (h *Holder) Forged(event string, body []byte) {
f, err := sdk.DecodeForge(body)
if event == forgeGroupEnd {
return
}
f, err := forgemsg.DecodeForge(body)
if err != nil {
h.Refuse(fmt.Errorf("a message of the forge bridge is not readable: %v", err))
return
}
switch {
case f.Kind == sdk.ForgeMerged:
case f.Kind == forgemsg.ForgeMerged:
h.ForgeMerged(f)
case f.Kind == sdk.ForgeRefused && f.Act == "merge":
case f.Kind == forgemsg.ForgeRefused && f.Act == "merge":
h.ForgeRefused(f)
}
}
// memberOfForge is the delivery a forge message is about: at the head it names.
func (h *Holder) memberOfForge(f sdk.Forge) *Delivery {
func (h *Holder) memberOfForge(f forgemsg.Forge) *Delivery {
if d := h.deliveries[IDOf(f.Repository, f.Head)]; d != nil && d.Number == f.Number {
return d
}
@@ -337,7 +348,7 @@ func (h *Holder) memberOfForge(f sdk.Forge) *Delivery {
// ForgeMerged takes the bridge's word that a pull request the gate handed to it merged: the merge is announced once,
// with the ask, its asker the teller and the heads (issue 1025), and the next member's turn comes at the next look.
func (h *Holder) ForgeMerged(f sdk.Forge) {
func (h *Holder) ForgeMerged(f forgemsg.Forge) {
h.mu.Lock()
defer h.mu.Unlock()
d := h.memberOfForge(f)
@@ -354,7 +365,7 @@ func (h *Holder) ForgeMerged(f sdk.Forge) {
// ForgeRefused takes the bridge's word that the forge refused a merge: nothing after it is merged, and the refusal is said
// as a stalled line (ADR 0441 §3, §4). The merge is handed again only at the next event that decides whether it may
// merge (mergeAgainNow).
func (h *Holder) ForgeRefused(f sdk.Forge) {
func (h *Holder) ForgeRefused(f forgemsg.Forge) {
h.mu.Lock()
defer h.mu.Unlock()
d := h.memberOfForge(f)
+1
View File
@@ -6,6 +6,7 @@ require (
git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient v0.0.0-20261104174344-ae5266e77639
git.novox.be/novox/mesh-contracts/go/seats/operatorchannel v0.0.0-20261104174344-ae5266e77639
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261104174359-05b3e2e1b94f
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg v0.0.0-20261104235927-c78941e80b59
github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0
)
+2
View File
@@ -4,6 +4,8 @@ git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient v0.0.0-202611
git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient v0.0.0-20261104174344-ae5266e77639/go.mod h1:Dj6fbL8Evgy72EWO3Gl3L3Ra8N/nJSzgEVwSGmeRam0=
git.novox.be/novox/mesh-contracts/go/seats/operatorchannel v0.0.0-20261104174344-ae5266e77639 h1:K/1yIkqAWKGtLEtJ0PibijIdpo5YATL/9IUVXFoEglM=
git.novox.be/novox/mesh-contracts/go/seats/operatorchannel v0.0.0-20261104174344-ae5266e77639/go.mod h1:osh0jQyNqYDM27tbxQo6HeUEXgLq9XUewKzNhUXadog=
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg v0.0.0-20261104235927-c78941e80b59 h1:pFb+lP0FxS3bF6N3gHKtjzyixI9b1vC/NLViT/lVzRQ=
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg v0.0.0-20261104235927-c78941e80b59/go.mod h1:Xxg9VI4y755tznAS2PygQFe2V3/aIzjRn3/06+Br9nQ=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261104174359-05b3e2e1b94f h1:jDsfDMd48MXY9Zershc3eRxnhmx2sSpUfQkLq5jAIds=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261104174359-05b3e2e1b94f/go.mod h1:08OomDXdwVN8/87nyDtv5R3rH/6LoP8SFFXz30bTYCM=
github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op h1:1BOWQJweNyvZMlpAHXGLiZQn9S+QXGcz3xh94lC0w6E=
+73
View File
@@ -1479,3 +1479,76 @@ func (h *Holder) sayClosed(key string, rec AskRecord, outcome string, words stri
h.Logf("[messenger] that the ask %s ended could not be said: %v", key, err)
}
}
// AsksQuery narrows what the tool `operator-channel.asks` answers (issue 595): Key one ask, State open (open and held) or
// ended, Since the asks opened at or after that time, Bound the most bytes of the answer's asks. The zero query keeps the
// whole answer, as before.
type AsksQuery struct {
Key, State string
Since time.Time
Bound int
}
// AsksAnswer is the answer of a narrowed query: the asks kept, how many the filters matched, and, when the bound cut the
// answer short, that it did and how.
type AsksAnswer struct {
Asks []AskView `json:"asks"`
Matched int `json:"matched"`
Cut string `json:"cut,omitempty"`
}
// AsksQueryOf reads the arguments of the tool; an empty set is the zero query.
func AsksQueryOf(a map[string]any) (AsksQuery, error) {
q := AsksQuery{Key: strArg(a, "key"), State: strArg(a, "state")}
if q.State != "" && q.State != "open" && q.State != "ended" {
return q, errorf("state is open or ended, not %q", q.State)
}
if since := strArg(a, "since"); since != "" {
t, err := time.Parse(time.RFC3339, since)
if err != nil {
return q, errorf("since is a time like 2026-11-04T22:00:00Z: %v", err)
}
q.Since = t
}
if v, ok := a["bound"].(float64); ok {
if v < 1 {
return q, errorf("bound is the most bytes of the answer, at least 1")
}
q.Bound = int(v)
}
return q, nil
}
// Narrow keeps of the views (open first, newest first) the asks the query names, and cuts the answer at its byte bound: the
// asks that fit, in order, and a said cut when one more did not.
func (q AsksQuery) Narrow(all []AskView) AsksAnswer {
var out AsksAnswer
out.Asks = []AskView{}
var kept []AskView
for _, v := range all {
open := v.State == StateOpen || v.State == StateHeld
switch {
case q.Key != "" && v.Key != q.Key,
q.State == "open" && !open,
q.State == "ended" && open,
!q.Since.IsZero() && v.Opened < q.Since.UTC().Format(time.RFC3339):
continue
}
kept = append(kept, v)
}
out.Matched = len(kept)
used := 0
for i, v := range kept {
if q.Bound > 0 {
raw, _ := json.Marshal(v)
if used+len(raw)+1 > q.Bound {
out.Cut = fmt.Sprintf("the answer is cut at its bound of %d bytes: %d of %d asks are shown, the %d after them are not "+
"(narrow with key, state or since, or raise bound)", q.Bound, i, len(kept), len(kept)-i)
break
}
used += len(raw) + 1
}
out.Asks = append(out.Asks, v)
}
return out
}
@@ -2,6 +2,7 @@ package main
import (
"encoding/json"
"reflect"
"strings"
"testing"
"time"
@@ -140,3 +141,55 @@ func TestAHeldAskNamesEveryKindAndWhyWhenNoChannelIsReady(t *testing.T) {
t.Fatalf("%q", why)
}
}
// Issue 595: operator-channel.asks answers one ask, the open or the ended, those opened from a time, and no more bytes than a
// bound, said when it cut the answer short; with no argument it keeps the whole answer.
func TestTheAsksVerbNarrowsItsAnswer(t *testing.T) {
view := func(key, state, opened string) AskView {
return AskView{Key: key, Asker: "mesh-delivery", Headline: "Approve " + key, State: state, Opened: opened}
}
all := []AskView{view("a.3", StateOpen, "2026-11-04T10:00:00Z"), view("a.2", StateHeld, "2026-11-03T10:00:00Z"),
view("a.1", "ended", "2026-11-02T10:00:00Z"), view("a.0", "ended", "2026-11-01T10:00:00Z")}
one, _ := json.Marshal(all[0])
for _, tc := range []struct {
name string
args map[string]any
want []string
matched int
cut bool
err bool
}{
{"no argument is the zero query", map[string]any{}, nil, 0, false, false},
{"key", map[string]any{"key": "a.1"}, []string{"a.1"}, 1, false, false},
{"state open takes open and held", map[string]any{"state": "open"}, []string{"a.3", "a.2"}, 2, false, false},
{"state ended", map[string]any{"state": "ended"}, []string{"a.1", "a.0"}, 2, false, false},
{"since", map[string]any{"since": "2026-11-02T12:00:00Z"}, []string{"a.3", "a.2"}, 2, false, false},
{"bound cuts and says so", map[string]any{"bound": float64(2*len(one) + 10), "state": "open"}, []string{"a.3", "a.2"}, 2, false, false},
{"a bound under the asks cuts", map[string]any{"bound": float64(len(one) + 5)}, []string{"a.3"}, 4, true, false},
{"a state that is neither", map[string]any{"state": "done"}, nil, 0, false, true},
{"a time that is not one", map[string]any{"since": "yesterday"}, nil, 0, false, true},
} {
q, err := AsksQueryOf(tc.args)
if (err != nil) != tc.err {
t.Errorf("%s: err = %v", tc.name, err)
continue
}
if tc.err {
continue
}
if tc.want == nil {
if q != (AsksQuery{}) {
t.Errorf("%s: %+v is not the zero query, which keeps today's answer", tc.name, q)
}
continue
}
got := q.Narrow(all)
var keys []string
for _, v := range got.Asks {
keys = append(keys, v.Key)
}
if !reflect.DeepEqual(keys, tc.want) || got.Matched != tc.matched || (got.Cut != "") != tc.cut {
t.Errorf("%s: asks %v matched %d cut %q, want %v, %d, cut %v", tc.name, keys, got.Matched, got.Cut, tc.want, tc.matched, tc.cut)
}
}
}
+18 -2
View File
@@ -735,8 +735,24 @@ func tools(h *Holder, l *listening) []stdio.Tool {
{Name: "operator-channel.asks",
Description: "The asks this router holds (novox/hq ADR 0259): open (or held, with since when and why no channel can carry it yet) first, then the last month's — each " +
"one's asker, headline, state, when it opened and expires, the channels it was offered on, how it " +
"ended (the warrant: who chose what, through which channel), and whether its asker was told.",
Run: func(map[string]any) (any, error) { return h.AsksNow(), nil }},
"ended (the warrant: who chose what, through which channel), and whether its asker was told. With no argument every " +
"ask, which is large; filter it: key for one ask, state open or ended, since (a time, RFC 3339) for the asks opened " +
"from then, and bound, the most bytes of the answer, which says when it cut it short.",
Input: map[string]any{
"key": str("one ask, by its key (asker.id); none for every ask"),
"state": str("open (open and held) or ended; none for both"),
"since": str("only the asks opened at or after this time, RFC 3339"),
"bound": map[string]any{"type": "integer", "description": "the most bytes of the asks in the answer; said when it cuts the answer short"}},
Run: func(a map[string]any) (any, error) {
q, err := AsksQueryOf(a)
if err != nil {
return nil, err
}
if q == (AsksQuery{}) {
return h.AsksNow(), nil
}
return q.Narrow(h.AsksNow()), nil
}},
{Name: "messenger_identities",
Description: "The accounts linked as the operator, one per channel kind: the account, the name its " +
"service shows, when it was linked, and from when it can approve (an hour after linking).",
+5 -3
View File
@@ -105,9 +105,11 @@ type Machine struct {
// ownerOf reads a file's owner from its inode; nil is the inode as the kernel says it. Tests set it
// so the folders they make stand for a non-root owner, whoever runs them.
ownerOf func(info os.FileInfo) (uid, gid int)
// clean is when a look below each real path last found no backup repository (LookKept).
clean map[string]time.Time
cleanMu sync.Mutex
// Beside runs a look below a share beside the pass; nil is a goroutine. Tests run it in place.
Beside func(func())
// looks is the last look below each real path (LookKept).
looks map[string]*lookBelow
looksMu sync.Mutex
// walk is the running pass's context, which ends the look below a share when the process stops.
walk context.Context
walkMu sync.Mutex
@@ -32,7 +32,7 @@ func newFake(t *testing.T) *fake {
}
func (f *fake) machine() *Machine {
return &Machine{Root: f.root, UID: f.uid, owners: f.owners, ownerOf: asTheLibrarysOwner, Now: func() time.Time { return f.now }, Run: func(_ context.Context, name string, args ...string) Ran {
return &Machine{Root: f.root, UID: f.uid, owners: f.owners, ownerOf: asTheLibrarysOwner, Beside: func(f func()) { f() }, Now: func() time.Time { return f.now }, Run: func(_ context.Context, name string, args ...string) Ran {
line := strings.Join(append([]string{name}, args...), " ")
f.ran = append(f.ran, line)
if (line == "exportfs -ra" || line == "sudo -n exportfs -ra") && f.onExportfs != nil {
+1 -2
View File
@@ -61,9 +61,8 @@ func (m *Machine) Pass(ctx context.Context, last *Status) Status {
if _, note := m.Backup(); note != "" {
s.Warnings = append(s.Warnings, note)
}
m.setWalk(ctx)
m.setWalk(ctx) // stays set: a look beside the pass ends when the process does
exports, refused, said := m.Judge(c, roster, wants)
m.setWalk(nil)
if ctx.Err() != nil {
// Stopping: write nothing and reload nothing, so Stopped starts at once.
s.Problems = append(s.Problems, "the pass was ended by the process stopping; nothing was written")
@@ -432,3 +432,89 @@ func TestAHangingCommandStillLetsStoppedRemoveTheFileInTime(t *testing.T) {
t.Errorf("the export file stays after an unassignment whose commands hung")
}
}
// Issue 1971: the look below a share runs beside the pass. A pass whose walk has not finished still writes
// its status, refuses the share until a look has finished, keeps believing a clean look while it is renewed,
// and a walk ended by the process stopping leaves no result.
func TestTheLookBelowRunsBesideThePassAndNeverHoldsItsStatus(t *testing.T) {
f := newFake(t)
f.mountinfo("/")
f.dir("/srv/clean/films/one")
f.file(ConfigFile, grantsConfig("clean=/srv/clean", "clean=ws:rw"))
f.roster("ws 10.0.0.3")
f.wants(map[string][]Want{"ws": {{Share: "clean", Access: "rw"}}})
m := f.machine()
var walks []func()
m.Beside = func(w func()) { walks = append(walks, w) } // the walk is still running
now := time.Date(2026, 11, 5, 5, 0, 0, 0, time.UTC)
m.Now = func() time.Time { return now }
s := m.Pass(context.Background(), nil)
if !s.At.Equal(now) || f.read(StatusFile) == "" || len(walks) != 1 {
t.Fatalf("the pass waited for the walk: at %v, walks %d", s.At, len(walks))
}
if !strings.Contains(refusedWhy(s, "clean"), "not been looked through yet") {
t.Errorf("a share whose look is unfinished is exported: %+v", s.Refused)
}
m.Pass(context.Background(), &s)
if len(walks) != 1 {
t.Errorf("a second walk started while one runs: %d", len(walks))
}
walks[0]()
if s = m.Pass(context.Background(), &s); refusedWhy(s, "clean") != "" {
t.Fatalf("the finished look was not used: %+v", s.Refused)
}
now = now.Add(LookKept)
s = m.Pass(context.Background(), &s)
if len(walks) != 2 || refusedWhy(s, "clean") != "" {
t.Errorf("the renewal must run beside the pass and the clean look stand: walks %d, %+v", len(walks), s.Refused)
}
ctx, cancel := context.WithCancel(context.Background())
m.Pass(ctx, &s)
cancel()
f.restic("/srv/clean/films/one/backup")
walks[1]()
if s = m.Pass(context.Background(), &s); refusedWhy(s, "clean") != "" {
t.Errorf("a walk ended by the stop left a result: %+v", s.Refused)
}
walks[len(walks)-1]()
if s = m.Pass(context.Background(), &s); !strings.Contains(refusedWhy(s, "clean"), "/srv/clean/films/one/backup") {
t.Errorf("the repository was not found by the next look: %+v", s.Refused)
}
}
// After a restart no look is remembered; a share the previous process's export file carries stays exported,
// and nothing is reloaded, while its first look runs (a delivery never unexports a mounted share).
func TestARestartKeepsTheExportedShareWhileItsFirstLookRuns(t *testing.T) {
f := newFake(t)
f.mountinfo("/", "/srv/clean")
f.dir("/srv/clean/films/one")
f.file(ConfigFile, grantsConfig("clean=/srv/clean", "clean=ws:rw"))
f.roster("ws 10.0.0.3")
f.wants(map[string][]Want{"ws": {{Share: "clean", Access: "rw"}}})
f.kernelReads()
first := f.machine()
s := first.Pass(context.Background(), nil)
before := f.read(ExportFile)
if refusedWhy(s, "clean") != "" || !strings.Contains(before, "/srv/clean ") {
t.Fatalf("the share was not exported to begin with: %+v %q", s.Refused, before)
}
m := f.machine()
var held []func()
m.Beside = func(w func()) { held = append(held, w) }
f.ran = nil
s = m.Pass(context.Background(), &s)
if len(held) != 1 || refusedWhy(s, "clean") != "" || f.read(ExportFile) != before || countExportfs(f.ran) != 0 {
t.Errorf("a restart with an unfinished walk unexported the share: %+v, ran %v", s.Refused, f.ran)
}
// The same name, a new folder holding a repository, the walk unfinished: refused, not carried by name.
f.restic("/srv/other/backup")
f.file(ConfigFile, grantsConfig("clean=/srv/other", "clean=ws:rw"))
m3 := f.machine()
m3.Beside = func(func()) {}
s = m3.Pass(context.Background(), &s)
if refusedWhy(s, "clean") == "" || strings.Contains(f.read(ExportFile), "/srv/other") {
t.Errorf("a share moved to a new folder was exported unlooked: %+v", s.Refused)
}
}
+80 -24
View File
@@ -244,33 +244,72 @@ const (
)
// LookKept is how long a look below a share that found no backup repository is believed. The look walks up
// to WalkLimit folders, which on a large library is too much for every 30 s pass; a repository put under a
// to WalkLimit folders, which on a large library takes minutes on a cold disk; a repository put under a
// share is found within this long. A look that found one, or was cut short, is never kept: the share stays
// refused and is looked at again on the next pass.
const LookKept = time.Hour
// repositoryBelow finds a backup repository within WalkDepth levels below dir. complete is false when
// the look stopped at its bound before it saw everything.
func (m *Machine) repositoryBelow(dir string) (found string, complete bool) {
now := time.Now()
if m.Now != nil {
now = m.Now()
}
m.cleanMu.Lock()
defer m.cleanMu.Unlock()
if at, kept := m.clean[dir]; kept && now.Sub(at) < LookKept {
return "", true
}
found, complete = m.walkBelow(dir)
if found == "" && complete {
if m.clean == nil {
m.clean = map[string]time.Time{}
// lookBelow is the last look below one folder, and whether another runs now.
type lookBelow struct {
found string
complete bool
at time.Time
done bool // a look has finished; until then there is no result
running bool
}
// repositoryBelow answers the last look below dir for a backup repository, and starts the next one beside
// the caller when there is none, or it was not clean, or LookKept has passed. The pass therefore never waits
// for the walk (issue 1971: a walk of minutes held the status's time back until the module was found
// unhealthy). pending is true while no look has finished yet; the share is refused until then, and while a
// kept clean look is being renewed the old result stands. complete is false when the look stopped at its
// bound or with the process.
func (m *Machine) repositoryBelow(dir string) (found string, complete, pending bool) {
m.looksMu.Lock()
r := m.looks[dir]
if r == nil {
if m.looks == nil {
m.looks = map[string]*lookBelow{}
}
m.clean[dir] = now
} else {
delete(m.clean, dir)
r = &lookBelow{}
m.looks[dir] = r
}
return found, complete
fresh := r.done && r.found == "" && r.complete && m.clock().Sub(r.at) < LookKept
start := !fresh && !r.running
if start {
r.running = true
}
m.looksMu.Unlock()
if start {
walk := func() {
f, c := m.walkBelow(dir)
m.looksMu.Lock()
defer m.looksMu.Unlock()
r.running = false
if w := m.walking(); w != nil && w.Err() != nil {
return // ended by the process stopping: no result, the old one stands
}
r.found, r.complete, r.at, r.done = f, c, m.clock(), true
}
if m.Beside != nil {
m.Beside(walk)
} else {
go walk()
}
}
m.looksMu.Lock()
defer m.looksMu.Unlock()
if !r.done {
return "", false, true
}
return r.found, r.complete, false
}
func (m *Machine) clock() time.Time {
if m.Now != nil {
return m.Now()
}
return time.Now()
}
// walkBelow is the look itself, breadth first, bounded by WalkDepth and WalkLimit.
@@ -342,7 +381,8 @@ func (m *Machine) Judge(c Config, roster map[string]string, wants map[string][]W
prefix, _ = netip.ParsePrefix(c.Range)
}
points := m.MountPoints()
backup, _ := m.Backup() // judged on every pass, never kept: see backup.go
onDisk, _ := m.Read(ExportFile) // what the previous process judged
backup, _ := m.Backup() // judged on every pass, never kept: see backup.go
backup.Paths = m.realPaths(backup.Paths)
shares := map[string]bool{}
for _, s := range c.Shares {
@@ -411,12 +451,16 @@ func (m *Machine) Judge(c Config, roster map[string]string, wants map[string][]W
no(fmt.Sprintf("The share %s's folder %s%s is inside the backup repository %s, and backups are never exported.", s.Name, real, via, inside))
continue
}
holds, complete := m.repositoryBelow(real)
holds, complete, pending := m.repositoryBelow(real)
if holds != "" {
no(fmt.Sprintf("The share %s's folder %s%s holds a backup repository (%s), and backups are never exported: share the folders beside it instead.", s.Name, real, via, holds))
continue
}
if !complete {
if pending && !carries(onDisk, real) {
no(fmt.Sprintf("The share %s's folder %s%s has not been looked through yet for a backup repository (the look runs beside the pass and takes a while on a large folder), so it is not exported yet.", s.Name, real, via))
continue
}
if !pending && !complete {
no(fmt.Sprintf("The share %s's folder %s%s holds more folders than the module looks through (%d levels, at most %d folders), so it could not make sure no backup repository is under it: share smaller folders.", s.Name, real, via, WalkDepth, m.limit()))
continue
}
@@ -487,6 +531,18 @@ func (m *Machine) Judge(c Config, roster map[string]string, wants map[string][]W
return out, refused, said
}
// carries says whether the export file already holds the line for this real folder (a share exported to no
// node has none and waits for its look; a share whose folder changed is a new folder). While a share's first look is unfinished (after a restart) it stays as the previous process judged
// it, so a delivery never unexports a share clients have mounted.
func carries(file, real string) bool {
for _, l := range strings.Split(file, "\n") {
if strings.HasPrefix(l, real+" ") {
return true
}
}
return false
}
func (m *Machine) limit() int {
if m.WalkLimit > 0 {
return m.WalkLimit