diff --git a/cmd/mesh-controller/operator.go b/cmd/mesh-controller/operator.go index 0bf944d..a2209f5 100644 --- a/cmd/mesh-controller/operator.go +++ b/cmd/mesh-controller/operator.go @@ -189,13 +189,17 @@ func readPrivateKey(path string) (string, error) { func personIssue(ctx context.Context, args []string) error { set := flag.NewFlagSet("operator issue", flag.ContinueOnError) invokes := set.String("invokes", "", "the tools this person may call, comma-separated, or * for every one") - if err := set.Parse(args); err != nil { + // Flags on either side of the name, because the usage this command prints puts them after it — + // and the standard parser stops at the first thing that is not a flag, so the order the command + // documents was the one order it refused (2026-09-28). + positionals, err := parseAround(set, args) + if err != nil { return err } - if set.NArg() != 1 { + if len(positionals) != 1 { return errors.New("operator issue --invokes ") } - name := set.Arg(0) + name := positionals[0] if *invokes == "" { return errors.New( "say what this person may call: --invokes mesh-catalog.catalog_tools,gitea.repo_create, " + diff --git a/cmd/mesh-controller/upgrades.go b/cmd/mesh-controller/upgrades.go index a720362..b058ecf 100644 --- a/cmd/mesh-controller/upgrades.go +++ b/cmd/mesh-controller/upgrades.go @@ -245,10 +245,12 @@ func (f following) SourceMoved(ctx context.Context, m link.SourceMoved) error { // commit it already records — so it is rebuilt and its record left alone. Writing this commit as // its source would make it permanently behind a repository its manifest does not come from. var from, packaging []inventory.Entry + already := 0 for _, e := range entries { switch { case sourceIs(e.Source, m): if e.Source.BuiltFrom == m.Commit { + already++ continue } // **A merge older than the last look at the source is history, not a move.** The forge @@ -264,6 +266,13 @@ func (f following) SourceMoved(ctx context.Context, m link.SourceMoved) error { } } if len(from) == 0 && len(packaging) == 0 { + // "Already built from it" and "nothing reads it" are different facts, and reading the first + // as the second sends somebody looking for a broken trigger when the mesh is up to date. + if already > 0 { + fmt.Printf("%s/%s merged into %s (%.8s); %d module(s) the mesh holds are already built "+ + "from it\n", m.Owner, m.Repo, m.Base, m.Commit, already) + return nil + } fmt.Printf("%s/%s merged into %s (%.8s); nothing the mesh holds reads it\n", m.Owner, m.Repo, m.Base, m.Commit) return nil diff --git a/internal/broker/broker.go b/internal/broker/broker.go index 93d5b54..c08ca29 100644 --- a/internal/broker/broker.go +++ b/internal/broker/broker.go @@ -66,6 +66,19 @@ func FromEnvironment() (Broker, error) { if err != nil { return Broker{}, err } + // **One bus, one address** (novox/hq ADR 0131). Everything the mesh hands out — a token, a + // machine's membership, a person's credential — must name the bus the control plane itself is + // connected to; the setting above predates the move and, on a mesh that has moved, still names + // the broker it moved from. The first person issued after the move was handed the retired + // broker's port and could not connect to anything (2026-09-28). + // + // Read from the credential rather than from a second setting somebody keeps in step: the + // control plane cannot be wrong about where it is connected. + if bus, on, err := OnNATS(); err == nil && on { + if where := strings.TrimPrefix(BareAddress(bus), "nats://"); where != "" { + address = where + } + } return Broker{Address: address, Fingerprint: fingerprint}, nil } diff --git a/internal/link/receive_nats.go b/internal/link/receive_nats.go index a4b21af..0d812b8 100644 --- a/internal/link/receive_nats.go +++ b/internal/link/receive_nats.go @@ -154,10 +154,47 @@ func (n *natsInbound) deliver(ctx context.Context, act func(context.Context, Con m.seq = meta.Sequence.Stream m.delivered = meta.NumDelivered } + // **Work that outlives the acknowledgement window says so while it runs.** + // + // The bus waits a fixed time to be told a message was taken, and then hands it to whoever + // consumes next — which is right for a consumer that died and wrong for one that is busy. + // Acting on a merge builds every module the merge changed: minutes of work against a + // thirty-second window. So the same merge was handed over again while the first build was + // still running, and again after that — on 2026-09-28 one merge ran the mesh's whole + // catalogue five times over and exhausted a public registry's pull limit. + // + // Here rather than in each handler, because the window belongs to the transport and every + // handler would otherwise have to remember it. It changes nothing about a handler that + // dies: a message is kept alive only while this goroutine is, so a controller that stops + // stops saying so, and the bus redelivers exactly as it should. + working := make(chan struct{}) + defer close(working) + go stillWorking(msg, working) } act(ctx, m) } +// heartbeatWhileWorking is how often a handler still running tells the bus so — comfortably inside +// the shortest acknowledgement window the mesh gives any of its consumers. +const heartbeatWhileWorking = 10 * time.Second + +// stillWorking keeps one message alive until the work on it returns. +// +// An error is not worth reporting: what the bus does when it is not told is redeliver, which is +// exactly what happens if this fails, and the handler's own outcome is the thing worth logging. +func stillWorking(msg *nats.Msg, done <-chan struct{}) { + tick := time.NewTicker(heartbeatWhileWorking) + defer tick.Stop() + for { + select { + case <-done: + return + case <-tick.C: + _ = msg.InProgress() + } + } +} + // kindOfSubject is how this transport's addressing becomes what the mesh calls a message. // // By subject, which is the only thing the server enforces: a body claiming to be a report does not diff --git a/internal/link/receive_nats_test.go b/internal/link/receive_nats_test.go index a10ed34..ad799e1 100644 --- a/internal/link/receive_nats_test.go +++ b/internal/link/receive_nats_test.go @@ -395,3 +395,60 @@ func TestEverySubjectTheControllerFollowsDecodesToAKind(t *testing.T) { t.Error("a subject nobody follows decoded to a kind") } } + +// **A handler slower than the acknowledgement window is not handed its message again.** +// +// The bus waits a fixed time to be told a message was taken and then redelivers, which is right for +// a consumer that died and wrong for one that is busy. Acting on a merge builds modules — minutes +// against a thirty-second window — and the same merge was handed over five times while the first +// build was still running (2026-09-28). Here the window is two seconds and the work takes six. +func TestNatsWorkSlowerThanTheWindowIsNotHandedOverAgain(t *testing.T) { + js := aBus(t) + // The controller's own consumer, with a window short enough to outlive in a test. + if err := js.EnsureConsumer(broker.Consumer{ + Name: broker.ControllerName, Stream: "CONTROL", Push: true, AckWaitSeconds: 2, + Why: "a window short enough to outlive in a test", + }); err != nil { + t.Fatal(err) + } + var mu sync.Mutex + handled := 0 + slow := make(chan struct{}) + s, stop := servingOn(t, js, nil) + defer stop() + s.listener = slowly{func() { + mu.Lock() + handled++ + first := handled == 1 + mu.Unlock() + if first { + time.Sleep(6 * time.Second) + close(slow) + } + }} + + body, err := json.Marshal(Report{Node: "anchor", Declared: "d1", Applied: []string{"store"}}) + if err != nil { + t.Fatal(err) + } + if _, err := js.Context().Publish(ReportSubject("anchor"), body); err != nil { + t.Fatal(err) + } + select { + case <-slow: + case <-time.After(30 * time.Second): + t.Fatal("the slow work never finished") + } + // A moment for a redelivery to arrive, if the bus were going to send one. + time.Sleep(3 * time.Second) + mu.Lock() + defer mu.Unlock() + if handled != 1 { + t.Fatalf("one report was handled %d times, so slow work is run again while it is running", handled) + } +} + +// slowly is a listener that runs whatever it was given. +type slowly struct{ work func() } + +func (s slowly) Heard(context.Context, Report) error { s.work(); return nil }