6 Commits

Include renames

Author	SHA1	Message	Date
Ettore Di Giacinto	016686a3db	chore(distributed): take the nats-io modules out of the build ... Distributed mode has not dialled a message broker since the control plane moved onto the workers' own outward tunnels and every fan-out family moved onto PostgreSQL LISTEN/NOTIFY. What was left was the dependency itself, and the code that existed only to feed it. Dropped from go.mod: nats-io/jwt/v2, nats-io/nats.go, nats-io/nkeys, nats-io/nuid and testcontainers-go/modules/nats, along with the fourteen indirect requires that only the NATS testcontainer pulled in. go.sum carries no nats line either, so the removal is not the partial kind where the require goes and the checksum stays. Deleted with them: pkg/natsauth in full, the broker client's remaining options and TLS files, the per-node JWT minting on both the register and the approve path, and the natsauth.Config parameter threaded through the node routes. The credential manager is renamed and stripped rather than deleted, because it still holds the tunnel token that every re-registration rotates. The bus flags stay accepted and ignored, and are now hidden, on every command that had them, so an existing unit file, compose file or Helm values file still starts on the day of the upgrade. What is not kept is the validation that REQUIRED one: a distributed frontend started with no bus URL is no longer fatal. The TLS paths lose type:"existingfile" deliberately, so a certificate deleted along with the broker cannot fail a startup. One operator-visible behaviour change: --nats-require-auth no longer makes an agent worker wait through admin approval. Ask for that wait with --distributed-require-auth, which already implied it. It is documented in the migration section and pinned from both sides. A deployment now needs PostgreSQL and the frontends' own HTTP listener, and nothing else. coverage-baseline.txt moves from 54.2 to 62.0. Assisted-by: Claude Opus 5 [claude-code] Signed-off-by: Ettore Di Giacinto <mudler@localai.io>	2026-09-20 03:05:35 +00:00
Ettore Di Giacinto	be14db21e2	feat(distributed): move the last nine fan-out families onto PostgreSQL ... Gallery progress and cancel, the operation cache's start and end, the model and backend cache invalidations, staging progress, and the prefix cache's observations and invalidations now travel on the LISTEN/NOTIFY carrier. No subject is published or subscribed on messaging.Client anywhere in the tree, which is what makes retiring that package a deletion rather than a migration: $ grep -rn 'natsClient\.Publish\|nats\.Publish\|\.Nats\.Publish\|QueueSubscribe\|SubscribeReply\|\.Request(' \ --include='*.go' core/ pkg/ | grep -v _test \ | grep -v 'c\.Request()\|ctx\.Request()\|Request()\.Context' \ | grep -v 'core/services/testutil/fakebus.go' core/services/messaging/client.go:168,170,172,227,234,236,250,252,254,268,269,287 core/services/messaging/interfaces.go:21,22,23 Every remaining hit is inside core/services/messaging itself. The production reads of the NATS client are now three, all of them the documented agent-worker exception: Close on shutdown, the agent pool's publisher, and the agent-cancel carrier passed to newFanoutBridges. Prefix-cache observations publish like every other family rather than through a method that refuses a message too large for a notification. The plan proposed such a refusal on the reasoning that a long prompt makes a chain of thousands of entries; ExtractChain caps a chain at Config.MaxDepth blocks, MaxDepth is a constant with no operator knob, and the chain reaching Sync.Observe has one source, the router's own extraction hook. A worst-case observation is a few kilobytes against an 8000-byte cap, so the hot-path spill the refusal was designed to avoid cannot occur, and shipping it would have added the programme's only deliberate message drop to guard a condition that cannot arise. pgbus gains FitsInline instead, a predicate that shares one size decision with Publish and decides nothing, and core/application refuses at startup to wire a prefix cache whose configured depth would put every observation over the cap. The carrier choice is no longer stated at four sites. StagingTracker.SetPublisher and SubscribeBroadcasts become one SetBroadcaster, so a tracker that publishes where its peers are not listening cannot be spelled; prefixcache.Sync gains SubscribeBroadcasts, which reads the carrier it publishes on; and the gallery service and the operation cache are wired by methods on DistributedServices that name no carrier at all, so the NATS client beside it cannot be handed over. OpCache.SetMessagingClient and GalleryService.SetNATSClient are renamed to SetBroadcaster so a missed call site fails to compile. Two pre-existing defects that the two-real-carrier specs surfaced are fixed. A progress tick published before a cancel and delivered after it cleared Cancelled and left the operation reading as still running on that replica; mergeStatus now drops a stale tick rather than merging it. GetStatus and GetAllStatus handed out the stored OpStatus pointer while the broadcast subscribers mutated it in place, so an /api/operations response could be marshalled mid-write; both now copy. Assisted-by: Claude Opus 5 [claude-code] Signed-off-by: Ettore Di Giacinto <mudler@localai.io>	2026-09-20 03:05:34 +00:00
Ettore Di Giacinto	c97a524013	refactor(distributed): move job and agent fan-out onto the PostgreSQL carrier ... Five of the six families whose subscriber is an open HTTP response rather than a process-lifetime cache now travel on pgbus: jobs.<id>.progress, jobs.<id>.result, jobs.<id>.cancel, agent.<name>.events.<user> and responses.<id>.cancel. Both ends of each move together, so there is no state where a publisher is on one carrier and its subscriber on the other. agent.<name>.cancel does NOT move, and the plan was wrong about why. Its only subscriber in the tree is the agent worker, which has no database and so cannot join the PostgreSQL carrier at all. Publishing that cancel on pgbus would have lost every cancel of a worker-run agent while returning nil, which reports a cancel that reached nobody as a cancel that was sent. EventBridge now names its cancel carrier separately, a frontend replica sets it to the carrier the worker reads, and it stays there until a cancel rides the worker's tunnel like every other verb addressed to a worker. The carrier drops at 256 rather than blocking, which is not safe on its own for a result: a lost result has no successor message. It is not the only path. The claiming replica persists the terminal line before it releases the claim, and an open progress stream re-reads the job row once after subscribing and then periodically, so a dropped terminal broadcast costs promptness and never the answer. Both per-request subscriptions close in a defer instead of on one return path, and pgbus grows Subscribers() so the leak they would otherwise cause can be asserted. It has no other symptom: only the first subscriber of a channel issues a LISTEN, so a leaked filter just adds one closure per notification for every stream the replica has ever served. Subscribe now issues its LISTEN before it registers, which makes that count a readiness signal rather than a figure to compare against itself. Two rules that were stated at several sites and pinned at none are now one each. The re-broadcaster is built beside the dispatcher and the bridge and handed to the dispatch loop, so no line is left that can point it at a carrier nobody subscribes to while every spec stays green. The set of statuses a job never leaves is one exported set that the SSE bridge and the store both read. The last hand-written subject filter in production code became messaging.SubjectAgentEventsWildcard. Assisted-by: Claude Opus 5 [claude-code] Signed-off-by: Ettore Di Giacinto <mudler@localai.io>	2026-09-20 03:05:34 +00:00
Ettore Di Giacinto	5562cd7ad3	feat(distributed): keep the listener draining and let it come back ... PostgreSQL holds undelivered notifications in a queue it shares with every session on the server, and it kills a listener that stops draining. Two failure modes follow, and both are silent: a carrier that blocked on a slow resolver would lose its connection and with it every later broadcast, and a carrier that reconnected without re-registering would be connected and deaf. The receive and dispatch halves were already separate. What was missing is everything around them. The listener path moves into listener.go and gains a carrier-level Dropped() so a replica that is behind can be seen; the queue depth and the spill retention become Config fields with exported defaults; the LISTEN session gets an application_name so an operator can count listeners in pg_stat_activity and a spec can drop exactly one of them; and OnReconnect fires after the re-LISTEN, on a goroutine of its own, because a callback re-hydrates from a database and must never run on the path whose only job is to drain. That callback is reached through an optional interface assertion, so deleting its invocation compiles and every adopter silently stops converging. The spec is the only guard, and it is named in a comment at the site. The slow consumer is proved through the transport rather than a seam: an ACCESS EXCLUSIVE lock on bus_messages stalls the resolver's spill SELECT for exactly as long as the spec holds it, and the listener is shown still draining and dropping while it does. The dropped connection is a pg_terminate_backend matched on the carrier's own application name. Neither Dropped nor IsConnected is on messaging.Broadcaster, and a spec asserts that over the interface type. Both are facts about a frontend; the conditions a scheduler acts on are facts about a worker, and no consumer holding the interface can read one as the other. Assisted-by: Claude Opus 5 [claude-code] Signed-off-by: Ettore Di Giacinto <mudler@localai.io>	2026-09-20 03:05:34 +00:00
Ettore Di Giacinto	ecfa96875a	fix(distributed): order LISTEN and UNLISTEN on one lock ... Unsubscribe decided a channel had lost its last subscriber under one lock and issued the UNLISTEN after releasing it. A Subscribe on the same root could decide to LISTEN in that window, and the two reached the connection in the wrong order: the root ended up not listened with a live subscription on it. It does not heal, because the next Subscribe sees the registration already there and never re-LISTENs, so the whole root stays deaf on that replica until the connection drops. The decision and the statement it implies now happen under one lock, held across both, at both call sites. A second lock and not the registration lock: issuing waits on the listener goroutine, delivery takes the registration lock, and holding that across the wait deadlocks the carrier. The race is spec'd through a barrier seam rather than by racing goroutines. The natural window is microseconds wide, and a spec that waits for it to open passes by luck; the seam scripts the interleaving, so the spec decides in both directions. Resolving a spilled message moved off the listener. PostgreSQL keeps undelivered notifications in a shared, fixed-size queue, so a listener that stops draining it can block COMMIT for every publisher on the server, not only this one. The listener now only drains; one resolver goroutine reads the row back and dispatches, which also keeps a spilled message and an inline one on the same subject in the order they were published. Three wiring lines that could be deleted with the suite staying green: the sweeper's start is now pinned by a Config interval, and the two lines that carry the bus into the deployment now refuse to boot when either is missing. A subscription can also report what it dropped, so the party that missed a message is the party that can see it. Assisted-by: Claude Opus 5 [claude-code] Signed-off-by: Ettore Di Giacinto <mudler@localai.io>	2026-09-20 03:05:34 +00:00
Ettore Di Giacinto	57cde52b8f	feat(distributed): carry fan-out on PostgreSQL LISTEN/NOTIFY ... Distributed mode needs an operator to run a NATS cluster. This adds the carrier that replaces its fan-out half, so a deployment eventually needs PostgreSQL and its own HTTP listener and nothing else. pgbus holds one PostgreSQL session per replica, pinned for the life of the process because LISTEN registrations belong to one backend session and a pooled handle would lose them on the next checkout. Publishes go out on the pool with pg_notify. Subjects map onto a channel by their first token, from a closed set of roots. A subject outside the set is refused at publish AND at subscribe rather than mapped to a channel of its own: a channel name is capped at 63 bytes, and one LISTEN per job id would be unbounded. Refused rather than dropped, because a subject that goes nowhere and reports nothing is the class of defect this work exists to remove. PostgreSQL refuses a notify payload of 8000 bytes or more, and several subjects on this bus exceed that in normal operation: a job result carries a whole LLM output, a gallery progress event carries one entry per node. Those are written to a row and the notification carries the id. What is measured against the cap is the ENCODED notification, not the caller's payload, because the subject and the envelope travel too. The filter grammar is not respelled here. Subscribe asks messaging.ValidFilter and delivery asks messaging.SubjectMatches, which makes this the first production caller of a matcher that had only test doubles. New refuses a DSN that names a different database from the pool: that pairing publishes successfully, delivers nothing, on every replica, and reports no error anywhere. Nothing publishes on it and nothing subscribes yet. The construction is wired anyway, because the DSN has exactly one legitimate source and a setting that decides whether any broadcast is delivered should not be invented by whichever call site is migrated first. Delivery is at-most-once, like NATS core. Nothing downstream may read a message it did not receive as evidence about a node: a carrier that cannot deliver is not a worker that is gone. Assisted-by: Claude Opus 5 [claude-code] Signed-off-by: Ettore Di Giacinto <mudler@localai.io>	2026-09-20 03:05:34 +00:00