feat(distributed): SyncedMap component + migrate finetune/quant/agent-tasks to cross-replica state#10542
Merged
Merged
Conversation
Introduce core/services/syncstate.SyncedMap[K,V]: a thread-safe in-memory map that keeps itself consistent across frontend replicas via NATS, with an optional pluggable durable Store and hydrate-from-source convergence. Several features keep process-local state surfaced to the API (finetune/quant jobs, agent tasks, model configs) and each hand-wired the same in-memory + NATS broadcast + read-through-store legs - or forgot to, reintroducing cross-replica staleness. SyncedMap makes that consistency a configuration choice: - local writes mutate the map, write through the Store, then broadcast a delta; - the apply path is memory-only and never re-publishes or re-writes the Store (structural echo-loop guard, mirroring galleryop.mergeStatus); - on Start and on NATS reconnect the map re-hydrates from the source (Store, else Loader); an optional periodic Reconcile repairs silent drift; - standalone mode (nil NATS client) is a strict in-memory no-op. Reconnect re-hydrate is wired via a new *messaging.Client.OnReconnect callback, consumed through an optional type-assertion so MessagingClient stays minimal. Adds messaging.SubjectSyncStateDelta and a reusable testutil.FakeBus (synchronous in-process MessagingClient with wildcard matching) for adopter tests. Component only; service migrations follow in subsequent commits. Signed-off-by: Ettore Di Giacinto <mudler@localai.io> Assisted-by: Claude:claude-opus-4-8 [Claude Code]
…tency FineTuneService kept jobs in a process-local map and, although it wrote them to Postgres, ListJobs/GetJob never read the store back and the wired natsClient was never used - so in distributed mode a job created on one replica was invisible to the others. Replace the map and the dead client with a syncstate.SyncedMap keyed by job ID, value *schema.FineTuneJob (the exact REST shape, so responses are unchanged). - Add a Store adapter (core/services/finetune/syncstore.go) over FineTuneStore, plus FineTuneStore.ListAll (global hydrate; per-user List kept) and an idempotent Upsert (create-or-update; Create alone fails on dup key). - Writes go through SyncedMap.Set/Delete (write-through + broadcast); reads use List/Get. The on-disk state.json path becomes the standalone Loader, keeping single-node restart recovery (stale->stopped / exporting->failed fixups). - Fold SetNATSClient/SetFineTuneStore into NewFineTuneService; app.go passes the distributed NATS client + store when distributed, nil otherwise. Signed-off-by: Ettore Di Giacinto <mudler@localai.io> Assisted-by: Claude:claude-opus-4-8 [Claude Code]
…a consistency AgentJobService.ListTasks read the process-local tasks map only, while ListJobs already read through the DB persister + dispatcher NATS - so in distributed mode a task created on one replica was invisible to the others. Back tasks with a syncstate.SyncedMap keyed by task ID (value schema.Task, the exact REST shape); jobs are left untouched. - Store adapter (task_syncstore.go) over the existing JobPersister (LoadTasks/SaveTask/DeleteTask); reads svc.persister/userID live so a persister swap needs no rebuild. No new persister methods required. - Task reads -> SyncedMap.List/Get; create/update -> Set (write-through + broadcast); delete -> Delete. The file persister now owns its own task set so the write-through path does not re-enter the SyncedMap lock (deadlock guard). - The distributed NATS client is not available at construction (start() precedes initDistributed), so it is injected via SetTaskSyncNATS, which rebuilds the still-empty map before Start/hydrate. Wired at the main, restart, and per-user (UserServicesManager) distributed sites. Signed-off-by: Ettore Di Giacinto <mudler@localai.io> Assisted-by: Claude:claude-opus-4-8 [Claude Code]
QuantizationService kept jobs in a process-local map persisted only to a local state.json, so in distributed mode jobs were neither visible across replicas nor durable cluster-wide. Back jobs with a syncstate.SyncedMap keyed by job ID (value *schema.QuantizationJob, the exact REST shape). - New distributed.QuantStore (GORM, table quantization_jobs) mirroring FineTuneStore: Create/Get/ListAll/Upsert(idempotent)/Delete, registered for AutoMigrate via distributed.InitStores (Stores.Quant). - New adapter (quantization/syncstore.go) over QuantStore implementing syncstate.Store, with record<->schema conversion. - Reads go through List/Get, writes through Set/Delete (write-through + broadcast); state.json is kept as the standalone Loader for single-node restart recovery (stale-job fixups preserved). - app.go passes the distributed NATS client + QuantStore when distributed, nil otherwise; Start/Close lifecycle mirrors finetune. Signed-off-by: Ettore Di Giacinto <mudler@localai.io> Assisted-by: Claude:claude-opus-4-8 [Claude Code]
![github-advanced-security[bot]](https://avatars.github.com/in/57789?s=60&v=4){kind=small}
gosec flagged the WithCancel in Start as "cancellation function not called" because the returned cancel is stored on the struct rather than called/deferred in scope. It is invoked in Close (covered by tests), and lifeCtx must outlive Start to drive the reconnect/reconcile goroutines. Suppress the verified false positive with a justified #nosec G118. Signed-off-by: Ettore Di Giacinto <mudler@localai.io> Assisted-by: Claude:claude-opus-4-8 [Claude Code]
…stgres Adds the real-infrastructure counterpart to the fake-bus unit tests, in the existing distributed e2e suite (testcontainers NATS + PostgreSQL). Two SyncedMap instances stand in for two frontend replicas - each with its OWN NATS connection to a shared server and a SHARED Postgres store (the distributed-mode invariant) - and assert, over the wire: - a create on replica A is observed by replica B; - an update and a delete propagate A -> B (delete prunes, which a reload cannot); - a late-joining replica recovers a job it never received a delta for, via store hydrate on Start (the at-most-once gap a fake bus cannot exercise); - a local Set is written through to the shared Postgres store. Signed-off-by: Ettore Di Giacinto <mudler@localai.io> Assisted-by: Claude:claude-opus-4-8 [Claude Code]
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Why
Follow-up to #10540. In distributed mode LocalAI runs multiple frontend replicas behind a round-robin LB. Several features keep process-local in-memory state surfaced to the API; each one that syncs correctly today (gallery statuses, staging progress, op cache) hand-wires the same three legs — in-memory map + NATS broadcast/apply + read-through store — and the ones that don't wire it silently serve stale/missing data on whichever replica didn't originate the change.
This introduces one well-tested component so cross-replica consistency is a configuration choice, then migrates the worst offenders onto it.
The component —
core/services/syncstate.SyncedMap[K,V]A thread-safe in-memory map that keeps itself consistent across replicas:
Set/Delete): mutate map → write-through optionalStore→ broadcast a{op,key,value}delta onstate.<name>.delta.OnApplyhook; never re-publishes, never re-writes the Store (structural echo-loop guard, same split asgalleryop.mergeStatus).Startand on NATS reconnect, re-hydrate from the source (Store, elseLoader); deltas apply in between; optional periodicReconcile. NATS core pub/sub is at-most-once, so hydrate-from-source is how a late/reconnecting replica catches up.nilNATS client): strict in-memory no-op — no broadcast/subscribe.Reconnect re-hydrate is wired via a new
*messaging.Client.OnReconnectcallback, consumed through an optional type-assertion so theMessagingClientinterface stays minimal. Addsmessaging.SubjectSyncStateDeltaand a reusabletestutil.FakeBus(synchronous in-processMessagingClientwith wildcard matching) for adopter tests.Migrations in this PR
core/services/finetune)ListJobs/GetJobnever read it back;natsClientwired but unusedSyncedMap[string,*schema.FineTuneJob]over aFineTuneStoreadapter (+ListAll, idempotentUpsert);state.jsonbecomes the standalone Loadercore/services/quantization)state.jsonSyncedMap[string,*schema.QuantizationJob]+ new durabledistributed.QuantStore(GORM, AutoMigrated);state.jsonkept as standalone Loadercore/services/agentpool)ListTasksread in-memory only (jobs already synced)SyncedMap[string,schema.Task]over the existingJobPersister; NATS injected viaSetTaskSyncNATSbefore hydrate at the main/restart/per-user sitesJobs in agentpool were already consistent (dispatcher NATS + DB read-through) and are left untouched. REST response shapes are unchanged in all three.
Deferred (separate PRs)
SyncedMap(dogfood;Loader=LoadModelConfigsFromPath,OnApply=ShutdownModel).Tests
syncstate: propagation, delete-prune, hydrate-on-start, Loader fallback, echo guard, Store write-through,OnApply, standalone no-op, reconnect re-hydrate.FakeBusconverge on create/update/delete; record↔schema round-trips; new GORM stores tested against real Postgres viatestutil.SetupTestDB.All written test-first.
go vet,gofmt, andgolangci-lint --new-from-merge-baseare clean acrosscore/...; agentpool suite passes under-race.Assisted-by: Claude:claude-opus-4-8 [Claude Code]