novox/mesh-catalog
18
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request 'The provisioner reads a provider's outcome per offer, and modules subscribe before they wait, so gitea's credentials, routes and early events are not lost (issue 774)' (#571) from fix/held-module-and-assigner-contract into main

Browse Source
This commit was merged in pull request #571.
This commit is contained in:
mesh-tools
2026-10-30 21:17:44 +00:00
parent 395a37ddda b5475cfa5c
commit 5d9cead336
13 changed files with 176 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/gitea/module.json
+2 -1
View File
@@ -423,7 +423,8 @@
{
"name": "forge-account",
"credential": {"form": "none"},
"scope": "mesh"
"scope": "mesh",
"answered-by": "helper"
}
],
"build": {
modules/mesh-config/cmd/mesh-config/main.go
+4 -2
View File
@@ -175,8 +175,8 @@ func resolving(c *Config) {
// reads the router's record of every waiting ask once a minute besides, so an answer heard nowhere is still taken.
func hearing(c *Config) {
go func() {
<-stdio.Served() // subscribed as soon as the runtime serves, never after a guessed wait: an event sent meanwhile is heard
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
time.Sleep(wait)
err := stdio.Subscribe(asks.Seat+".decided."+Module, func(e stdio.Envelope) error {
var w asks.Warrant
if json.Unmarshal(e.Body, &w) == nil {
@@ -191,12 +191,13 @@ func hearing(c *Config) {
return
}
logf("[mesh-config] does not hear the operator's answers yet (%v); asking again in %s", err, wait)
time.Sleep(wait)
}
}()
for _, pattern := range []string{"gitea.pull.merged", "gitea.pull.closed", "gitea.forge.pull-request.closed"} {
go func(pattern string) {
<-stdio.Served() // subscribed as soon as the runtime serves, never after a guessed wait: an event sent meanwhile is heard
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
time.Sleep(wait)
err := stdio.Subscribe(pattern, func(e stdio.Envelope) error {
var p PullEvent
if json.Unmarshal(e.Body, &p) != nil {
@@ -212,6 +213,7 @@ func hearing(c *Config) {
return
}
logf("[mesh-config] does not hear %s yet (%v); asking again in %s", pattern, err, wait)
time.Sleep(wait)
}
}(pattern)
}
modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go
+1 -1
View File
@@ -91,7 +91,7 @@ func main() {
// events from then on (novox/hq ADR 0406 §1, issue 975): no tick, no forge poll, no reconcile loop. Each step is retried,
// each failure said, never given up on quietly.
func run(h *Holder, l *listening) {
time.Sleep(500 * time.Millisecond) // Serve first: the state is reached through it
<-stdio.Served() // Serve first: the state is reached through it, and an event is heard from as soon as it serves
loadKept(h, l)
go h.Work()
for _, pattern := range followed {
modules/mesh-healers/cmd/mesh-healers/main.go
+4 -2
View File
@@ -74,8 +74,8 @@ func main() {
})
// S15 over both logs: the seat's act-recorded on a durable consumer, beside the controller's bucket's watch.
go func() {
<-stdio.Served() // subscribed as soon as the runtime serves, never after a guessed wait: an event sent meanwhile is heard
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
time.Sleep(wait)
err := stdio.Subscribe(RecordedEvent, func(e stdio.Envelope) error {
h.ActRecorded(e.Body)
return nil
@@ -84,18 +84,20 @@ func main() {
return
}
logf("[%s] does not hear %s yet (%v); asking again in %s", Module, RecordedEvent, err, wait)
time.Sleep(wait)
}
}()
for _, event := range []string{RaisedEvent, ChangedEvent} {
go func(event string) {
<-stdio.Served() // subscribed as soon as the runtime serves, never after a guessed wait: an event sent meanwhile is heard
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
time.Sleep(wait)
err := stdio.Subscribe(event, func(e stdio.Envelope) error { return h.Heard(e.Body) })
if err == nil {
logf("[%s] hears %s", Module, event)
return
}
logf("[%s] does not hear %s yet (%v); asking again in %s", Module, event, err, wait)
time.Sleep(wait)
}
}(event)
}
modules/mesh-healers/cmd/mesh-healers/whole_integration_test.go
+24
View File
@@ -51,6 +51,21 @@ type player struct {
next int
pending map[string]chan wire
asked []string
// subscribed are the patterns the bundle subscribed, as the runtime takes them: an event is delivered only once its
// pattern is taken, as the runtime's consumer would hold it until then.
subscribed []string
}
// hears says the bundle has subscribed pattern.
func (p *player) hears(pattern string) bool {
p.mu.Lock()
defer p.mu.Unlock()
for _, s := range p.subscribed {
if s == pattern {
return true
}
}
return false
}
func (p *player) send(m map[string]any) {
@@ -198,7 +213,14 @@ func (p *player) handle(m wire) {
p.send(map[string]any{"id": id, "method": "mesh/state", "params": map[string]any{"watch": watch, "change": change}})
}
case "mesh/subscribe":
var sub struct {
Pattern string `json:"pattern"`
}
_ = json.Unmarshal(m.Params, &sub)
p.answer(m.ID, map[string]any{}, nil)
p.mu.Lock()
p.subscribed = append(p.subscribed, sub.Pattern)
p.mu.Unlock()
case "mesh/state.keys":
kv, err := p.kv(a.State)
if err != nil {
@@ -368,6 +390,8 @@ func TestTheHealersWholeActOnTheBusAndBrake(t *testing.T) {
return a != nil && strings.Contains(fmt.Sprint(a["why"]), "names a controller")
})
// The runtime's consumer holds an event until the bundle subscribes: delivered once it hears them.
eventually(t, "the healers hearing the keeper's conditions", func() bool { return p.hears(RaisedEvent) && p.hears(ChangedEvent) })
// The controller holds the lease: the induced failure is written as would, never acted on.
p.deliver(RaisedEvent, raised("bus.EVENTS.records.missing", "records"))
eventually(t, "the would-act written", func() bool { return len(keys(Module+"_"+WouldState)) == 1 })
modules/mesh-issues/cmd/mesh-issues/main.go
+2 -1
View File
@@ -163,14 +163,15 @@ func hearOpened(t *Tracker) {
}
hearings = append(hearings, hearing{DeliveredEvent, t.OnDelivered})
for _, s := range hearings {
<-stdio.Served() // subscribed as soon as the runtime serves, never after a guessed wait: an event sent meanwhile is heard
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
time.Sleep(wait)
err := stdio.Subscribe(s.pattern, s.handle)
if err == nil {
logf("[mesh-issues] the triage hears %s", s.pattern)
break
}
logf("[mesh-issues] the triage does not hear %s yet (%v); asking again in %s", s.pattern, err, wait)
time.Sleep(wait)
}
}
}
modules/mesh-knowledge/cmd/mesh-knowledge/main.go
+1 -1
View File
@@ -114,7 +114,6 @@ func callerOf(args map[string]any) string { return stdio.CallerOf(args) }
// hear subscribes to the router's word on this module's asks, asked again until the runtime takes it.
func hear(k *KB) {
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
time.Sleep(wait)
err := stdio.Subscribe(asks.Seat+".decided."+Module, func(e stdio.Envelope) error {
var w asks.Warrant
if json.Unmarshal(e.Body, &w) == nil {
@@ -127,6 +126,7 @@ func hear(k *KB) {
return
}
logf("[mesh-knowledge] does not hear the operator's answers yet (%v); asking again in %s", err, wait)
time.Sleep(wait)
}
}
modules/mesh-mail/cmd/mesh-mail/main.go
+2 -1
View File
@@ -250,8 +250,8 @@ func waitsTellTicker(b *Bundle) {
// approved send goes out as soon as the operator answers it, never only when something later calls
// mail_send_status (novox/hq ADR 0361 §2, review round 1 of issue 790).
func hear(b *Bundle) {
<-stdio.Served() // subscribed as soon as the runtime serves, never after a guessed wait: an event sent meanwhile is heard
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
time.Sleep(wait)
err := stdio.Subscribe(asks.Seat+".decided."+Module, func(e stdio.Envelope) error {
var w asks.Warrant
if json.Unmarshal(e.Body, &w) == nil {
@@ -264,6 +264,7 @@ func hear(b *Bundle) {
return
}
logf("[mesh-mail] does not hear the operator's answers yet (%v); asking again in %s", err, wait)
time.Sleep(wait)
}
}
modules/mesh-provisioner/cmd/mesh-provisioner/holder.go
+71 -32
View File
@@ -66,8 +66,10 @@ const (
// BoundReads are the providers' entries this module's manifest reads, `<module>.bound`, whose changes it hears. Which
// path judges a provider is its manifest's word (keepsEntries), not this list: a provider that declares `bound` and is
// not read here leaves its keys asked, saying its entries are not read.
var BoundReads = []string{"gitea", "keycloak", "postgres", "route-adapter", "mesh-vault"}
// not read here leaves its keys asked, saying its entries are not read. route-proxy (mesh-control) and node-tools
// (mesh-tools) answer route and mcp-endpoint by the provider helper from outside this catalogue (issue 774, sweep of
// 2026-10-30 20:10: their keys stood asked on every node).
var BoundReads = []string{"gitea", "keycloak", "postgres", "route-adapter", "route-proxy", "node-tools", "mesh-vault"}
// boundRead is a provider's entries as this module reads them.
func boundRead(module string) string { return module + ".bound" }
@@ -546,7 +548,7 @@ type entry struct {
// asked, said so.
func (h *Holder) outcome(key string, p *provisions.Provision) {
p.Ask, p.Warrant = "", nil
keeps, err := h.keepsEntries(p.Provider.Module)
keeps, err := h.keepsEntries(p.Provider.Module, p.Provision)
if err != nil {
p.State, p.Why = provisions.Asked, "the manifest of "+p.Provider.Module+" cannot be read: "+err.Error()
return
@@ -644,39 +646,30 @@ func (h *Holder) haveOutcome(p *provisions.Provision) {
}
}
// keepsEntries says a provider's manifest at the trunk declares the state `bound`: its outcome per consumer is its
// entries there; any other provider's is its own have.
func (h *Holder) keepsEntries(module string) (bool, error) {
mk, err := delivery.ModulesKey(module)
if err != nil {
return false, err
}
e, err := h.Store.Get(ModulesRead, mk)
if err != nil || e == nil {
return false, fmt.Errorf("the indexer holds no modules.%s (%v)", module, err)
}
m, err := delivery.DecodeModule(e.Value)
if err != nil || m.Manifest == "" {
return false, fmt.Errorf("the indexer's modules.%s names no manifest", module)
}
key, err := delivery.ManifestsKey(m.Manifest)
if err != nil {
return false, err
}
me, err := h.Store.Get(ManifestsRead, key)
if err != nil || me == nil {
return false, fmt.Errorf("the indexer holds no manifest of %s (%v)", module, err)
}
man, err := delivery.DecodeManifest(me.Value)
// keepsEntries says whether a provider's outcome for one provision is its entries (its manifest at the trunk declares the
// state `bound`) rather than its own have. When any offer of the provider says `answered-by: helper`, only those offers
// are answered by its entries: a module running the helper for one provision and making another some other way (gitea:
// the helper answers forge-account, the vault issues gitea-read, the package harness makes npm-package-registry) is
// judged per offer, so the helper's "does not serve" for an offer it does not run never becomes the provision's outcome
// (issue 774, sweep of 2026-10-30 20:10). A provider whose offers say nothing keeps every outcome in its entries.
func (h *Holder) keepsEntries(module, provision string) (bool, error) {
man, err := h.providerManifest(module)
if err != nil {
return false, err
}
return keepsEntriesFor(man, provision)
}
// keepsEntriesFor is keepsEntries over a provider's manifest JSON.
func keepsEntriesFor(manifest, provision string) (bool, error) {
var decl struct {
State []json.RawMessage `json:"state"`
State []json.RawMessage `json:"state"`
Provides []json.RawMessage `json:"provides"`
}
if err := json.Unmarshal([]byte(man.JSON), &decl); err != nil {
if err := json.Unmarshal([]byte(manifest), &decl); err != nil {
return false, err
}
bound := false
for _, raw := range decl.State {
var name string
if json.Unmarshal(raw, &name) != nil {
@@ -687,10 +680,56 @@ func (h *Holder) keepsEntries(module string) (bool, error) {
name = full.Name
}
if name == "bound" {
return true, nil
bound = true
}
}
return false, nil
if !bound {
return false, nil
}
helped, says := false, false
for _, raw := range decl.Provides {
var full struct {
Name string `json:"name"`
AnsweredBy string `json:"answered-by"`
}
if json.Unmarshal(raw, &full) != nil || full.AnsweredBy != catalogue.AnsweredByHelper {
continue
}
says = true
if full.Name == provision {
helped = true
}
}
return helped || !says, nil
}
// providerManifest is a provider's manifest JSON at the trunk, as the indexer holds it.
func (h *Holder) providerManifest(module string) (string, error) {
mk, err := delivery.ModulesKey(module)
if err != nil {
return "", err
}
e, err := h.Store.Get(ModulesRead, mk)
if err != nil || e == nil {
return "", fmt.Errorf("the indexer holds no modules.%s (%v)", module, err)
}
m, err := delivery.DecodeModule(e.Value)
if err != nil || m.Manifest == "" {
return "", fmt.Errorf("the indexer's modules.%s names no manifest", module)
}
key, err := delivery.ManifestsKey(m.Manifest)
if err != nil {
return "", err
}
me, err := h.Store.Get(ManifestsRead, key)
if err != nil || me == nil {
return "", fmt.Errorf("the indexer holds no manifest of %s (%v)", module, err)
}
man, err := delivery.DecodeManifest(me.Value)
if err != nil {
return "", err
}
return man.JSON, nil
}
// retired says a retiring key the operator approved is retired: its provider's entry is gone.
@@ -698,7 +737,7 @@ func (h *Holder) retired(key string, p provisions.Provision) (bool, string) {
if p.Warrant == nil || p.Warrant.Option != OptionApprove {
return false, ""
}
if keeps, err := h.keepsEntries(p.Provider.Module); err != nil || !keeps {
if keeps, err := h.keepsEntries(p.Provider.Module, p.Provision); err != nil || !keeps {
return false, ""
}
e, err := h.Store.Get(boundRead(p.Provider.Module), key)
modules/mesh-provisioner/cmd/mesh-provisioner/holder_test.go
+60
View File
@@ -412,3 +412,63 @@ func TestTheHolderHoldsNoTimer(t *testing.T) {
})
}
}
// manifestOf puts a provider's manifest into the indexer's reads as the holder finds it.
func (m *mesh) manifestOf(module, raw string) {
digest := delivery.ManifestDigest([]byte(raw))
value, _ := delivery.EncodeManifest(delivery.Manifest{Module: module, JSON: raw})
_ = m.s.Put(ManifestsRead, mustKey(delivery.ManifestsKey(digest)), json.RawMessage(value))
_ = m.s.Put(ModulesRead, mustKey(delivery.ModulesKey(module)),
json.RawMessage(`{"v":1,"none2":"a test","manifest":"`+digest+`","repository":"novox/mesh-catalog","commit":"c1"}`))
}
// A provider's outcome is read per offer: from its entries where the provider helper answers that offer, from its have
// where it does not, and the entries of route-proxy and node-tools are read (issue 774, the sweep of 2026-10-30 20:10:
// <node>.claude-code.gitea-read failed "gitea does not serve" on all four nodes while gitea's helper answers only
// forge-account, and novox.gitea.route and <node>.claude-code.mcp-endpoint stood asked, their providers' entries unread).
func TestAProvidersOutcomeIsReadPerOffer(t *testing.T) {
const (
gitea = `{"module":"gitea","version":"1","state":["bound","announced"],"provides":[{"name":"npm-package-registry","credential":{"form":"pair"}},{"name":"gitea-read","credential":{"issued-by":"software","tool":"gitea_agent_token"}},{"name":"forge-account","credential":{"form":"none"},"answered-by":"helper"}]}`
routes = `{"module":"route-proxy","version":"1","state":["bound"],"provides":[{"name":"route","scope":"mesh","identity":false,"answered-by":"helper"}]}`
nodeTools = `{"module":"node-tools","version":"1","state":["bound"],"provides":[{"name":"mcp-endpoint","scope":"node","answered-by":"helper"}]}`
running = `{"v":1,"fp":"src2:` + "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd" + `","phase":"running","health":"ok"}`
refused = `{"state":"failed","refused":"claude-code on ace is bound to gitea on novox for gitea-read, which gitea does not serve"}`
)
for _, row := range []struct {
name string
node string
plan map[string]catalogue.Plan
key string
bucket string // the provider's entries, written as entry
entry string
have string // the provider's have key, written running
state string
}{
{"gitea-read: the helper's refusal is not its outcome; gitea runs, so provided", "ace",
map[string]catalogue.Plan{"claude-code": planOf(catalogue.PlanProvision{Provision: "gitea-read", Provider: "novox", ProviderModule: "gitea"})},
"ace.claude-code.gitea-read", "gitea.bound", refused, "novox.gitea", provisions.Provided},
{"forge-account: still the helper's entry", "ace",
map[string]catalogue.Plan{"mesh-build-go": planOf(catalogue.PlanProvision{Provision: "forge-account", Provider: "novox", ProviderModule: "gitea"})},
"ace.mesh-build-go.forge-account", "gitea.bound", `{"state":"provided"}`, "", provisions.Provided},
{"route: route-proxy's entry is read", "novox",
map[string]catalogue.Plan{"gitea": planOf(catalogue.PlanProvision{Provision: "route", Provider: "novox", ProviderModule: "route-proxy"})},
"novox.gitea.route", "route-proxy.bound", `{"state":"provided"}`, "", provisions.Provided},
{"mcp-endpoint: node-tools' entry is read", "ace",
map[string]catalogue.Plan{"claude-code": planOf(catalogue.PlanProvision{Provision: "mcp-endpoint", Provider: "ace", ProviderModule: "node-tools"})},
"ace.claude-code.mcp-endpoint", "node-tools.bound", `{"state":"provided"}`, "", provisions.Provided},
} {
m := newMesh()
m.manifestOf("gitea", gitea)
m.manifestOf("route-proxy", routes)
m.manifestOf("node-tools", nodeTools)
m.atWant[row.node], m.atNewest[row.node] = row.plan, row.plan
m.s.write(t, row.bucket, row.key, row.entry)
if row.have != "" {
m.s.write(t, HaveRead, row.have, running)
}
m.compose(t, row.node)
if p := m.state(t, row.key); p.State != row.state {
t.Errorf("%s: %s is %s (%s)", row.name, row.key, p.State, p.Why)
}
}
}
modules/mesh-provisioner/cmd/mesh-provisioner/main.go
+2 -1
View File
@@ -136,8 +136,8 @@ func main() {
return nil
})
go func() {
<-stdio.Served() // subscribed as soon as the runtime serves, never after a guessed wait: an event sent meanwhile is heard
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
time.Sleep(wait)
err := stdio.Subscribe(asks.Seat+".decided."+Module, func(stdio.Envelope) error {
for _, said := range h.Decide() {
logf("[%s] %s", Module, said)
@@ -148,6 +148,7 @@ func main() {
return
}
logf("[%s] does not hear the operator's answers yet (%v); asking again in %s", Module, err, wait)
time.Sleep(wait)
}
}()
if err := stdio.Serve("", append(Tools(s), holderTools(h)...)); err != nil {
modules/mesh-provisioner/module.json
+2
View File
@@ -26,6 +26,8 @@
"keycloak.bound",
"postgres.bound",
"route-adapter.bound",
"route-proxy.bound",
"node-tools.bound",
"mesh-vault.bound"
],
"tools": [
modules/mesh-watcher/cmd/mesh-watcher/main.go
+1 -1
View File
@@ -114,7 +114,7 @@ func run(w *Watcher, l *listening, node string) {
w.Heartbeat(b)
return nil
}
time.Sleep(500 * time.Millisecond)
<-stdio.Served() // subscribed as soon as the runtime serves, never after a guessed wait: a heartbeat sent meanwhile is heard
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
err := stdio.Subscribe(ControllerSeat+".*", handle)
if err == nil {