mirror of
https://github.com/mudler/LocalAI.git
synced 2026-09-21 05:34:56 -04:00
Ettore Di Giacinto
78f4ff7934
The three NATS queue groups jobs.new, jobs.mcp-ci.new and agent.execute are gone. Dispatching work is now a row in a work_claims table, taken by one frontend replica with SELECT ... FOR UPDATE SKIP LOCKED and driven on an agent worker as a streaming control RPC over that worker's tunnel. Exactly-one delivery among competing consumers is a database problem, not a broker feature. An agent worker has no database, so it never claims; it executes what the claiming replica hands it. A claim must not outlive the replica that took it. The reap releases a claim whose owner is no longer a live replica in the instances table, on the database clock, and never asks how long the claim has been held. A job that legitimately runs for an hour on a heartbeating replica is left alone, while a claim whose owner stopped heartbeating becomes claimable again within one liveness window. A replica with no advertised address has no instances row at all, so it refuses to claim rather than have its work reaped out from under it mid-run. The settle rule is stated once, in settleClaim, and every exit path calls it. A transport failure releases the claim and never completes or discards it; only a decoded reply line completes it. That line is deliberately not cluster.IsWorkerAnswer, which accepts the stream refusals a worker's tunnel writes before any request body reaches its control server: completing on those would discard work that never ran. The terminal line is persisted before the claim is completed, so a store that refuses leaves the claim standing rather than leaving the job running for ever. That is the dropped-result defect fixed structurally rather than by retry. This also surfaces a pre-existing gap rather than causing one: no worker has ever served plain task jobs, and publishing them into an empty queue group left them running with no trace. Such a claim is now failed with a reason. Removes QueueWorkers, --agent-subject and --agent-queue, and narrows an agent worker's minted JWT by agent.execute and jobs.mcp-ci.new. Assisted-by: Claude Opus 5 [claude-code] Signed-off-by: Ettore Di Giacinto <mudler@localai.io>
71 lines
3.0 KiB
Go
71 lines
3.0 KiB
Go
// SPDX-License-Identifier: MIT
|
|
|
|
package application
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
|
|
"github.com/mudler/LocalAI/core/config"
|
|
"github.com/mudler/LocalAI/core/services/jobs"
|
|
"github.com/mudler/LocalAI/core/services/messaging"
|
|
"github.com/mudler/LocalAI/core/services/nodes"
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
// startJobDispatchLoop builds AND STARTS the loop that takes queued work off the
|
|
// job store and drives it on an agent worker.
|
|
//
|
|
// One function rather than a construction here and a Start somewhere else, and
|
|
// that is the point rather than tidiness. A loop that is built and never
|
|
// started is a replica that writes claim rows and takes none, so every job in
|
|
// the deployment is accepted and none is ever run, and nothing anywhere says
|
|
// so. As a separate statement in the start-up path that line's loss has no
|
|
// symptom and no spec reaches it: initDistributed opens a database and a bus.
|
|
// Fused here, the loop cannot exist without running.
|
|
//
|
|
// The rest is a named function for the reason newAgentControl is one: two of
|
|
// these arguments are silent when they are wrong.
|
|
//
|
|
// The BROADCASTER is the one worth naming. It is what re-publishes the progress
|
|
// and result lines a worker asks for, and it is checked against the allow list
|
|
// for that worker's node type. A loop built without one dispatches work
|
|
// perfectly well and every SSE stream in the deployment goes quiet: the job
|
|
// runs, the answer is persisted, and the user watching it sees nothing until
|
|
// they reload. That is a whole feature lost to a nil field, with no error
|
|
// anywhere, so it is refused here.
|
|
//
|
|
// The SELECTOR is built here rather than borrowed from newAgentControl, and
|
|
// deliberately: nodes.AgentSelector holds no per-caller state, and sharing one
|
|
// would couple the dispatch loop's lifetime to MCP's for nothing.
|
|
func startJobDispatchLoop(ctx context.Context, cfg config.DistributedConfig, db *gorm.DB, store *jobs.JobStore,
|
|
registry *nodes.NodeRegistry, conns nodes.AgentConnectionReader,
|
|
control *nodes.ControlClient, bus messaging.Broadcaster) (*jobs.DispatchLoop, error) {
|
|
if cfg.InstanceID == "" {
|
|
return nil, fmt.Errorf("the job dispatch loop was built with no instance id: its claims could not be told from ones a dead replica left")
|
|
}
|
|
if registry == nil || conns == nil {
|
|
return nil, fmt.Errorf("the job dispatch loop was built with no way to find a connected agent worker")
|
|
}
|
|
if bus == nil {
|
|
return nil, fmt.Errorf("the job dispatch loop was built with no broadcaster: every job would run with its progress and its result reaching no SSE stream in the deployment")
|
|
}
|
|
loop, err := jobs.NewDispatchLoop(jobs.DispatchConfig{
|
|
DB: db,
|
|
Owner: cfg.InstanceID,
|
|
Selector: nodes.NewAgentSelector(registry, conns, cfg.InstanceID),
|
|
Control: control,
|
|
// The allow list lives in nodes and is keyed on the worker's node type;
|
|
// nothing here decides what a worker may broadcast on.
|
|
Broadcast: nodes.NewRebroadcaster(bus),
|
|
Store: store,
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if err := loop.Start(ctx); err != nil {
|
|
return nil, err
|
|
}
|
|
return loop, nil
|
|
}
|