From 225da13860bca3b6655a8413a9d97ef7f692f84c Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Mon, 20 Jul 2026 20:55:50 +0000 Subject: [PATCH] fix(openresponses): make responses visible and cancellable across replicas In distributed mode the Open Responses store is process-local: a sync.OnceValue over a map behind an RWMutex. With several frontend replicas behind a round-robin load balancer, every request that lands on a replica other than the creator misses. Measured on a live 2-replica cluster (#10993): the same response id returns 200 on the creating replica and 404 on its peer, and a cancel on the peer returns 404 without ever invoking CancelFunc, so generation runs to completion on the other replica while the caller is told the response does not exist. previous_response_id chaining fails through the same lookup. Split the state by what can actually cross a process boundary: - Replicated: response metadata (request, response resource, owner, expiry, stream/background flags) via syncstate.SyncedMap, the same component finetune, quantization and agent tasks already use. A local miss in Get/FindItem now falls back to it and returns a read-only remote view, so polling and chaining resolve on any replica. - Delegated: cancellation. context.CancelFunc is a function pointer and exists only in the creating process, so a cancel that lands elsewhere is broadcast on responses..cancel and applied by whichever replica holds the function. The broadcast is fire-and-forget rather than request/reply: if the owner crashed or was scaled down nobody answers, and the handler must not block on a reply that will never come. The replicated status moves to cancelled either way, which is truthful, since a dead owner's generation died with its process. - Refused: streaming resume. The resume buffer is a byte log plus a live notification channel and cannot be replicated without shipping every token over the bus. A resume that reaches the wrong replica now returns HTTP 409 naming the owning replica via the new ErrResponseNotLocal, instead of an empty event list that looks like a finished stream. It is deliberately distinct from ErrOffsetLost, which means the owner's buffer evicted the requested events. Standalone deployments never call EnableDistributed and keep exactly the previous process-local behaviour. Fixes #10993 Signed-off-by: Ettore Di Giacinto Assisted-by: Claude Code:claude-opus-4-8[1m] [Read] [Edit] [Bash] --- .../http/endpoints/openresponses/responses.go | 15 + core/http/endpoints/openresponses/store.go | 193 ++++++++-- core/http/endpoints/openresponses/sync.go | 344 ++++++++++++++++++ .../http/endpoints/openresponses/sync_test.go | 208 +++++++++++ core/http/routes/openresponses.go | 11 + core/services/messaging/subjects.go | 13 + docs/content/features/text-generation.md | 16 + 7 files changed, 770 insertions(+), 30 deletions(-) create mode 100644 core/http/endpoints/openresponses/sync.go create mode 100644 core/http/endpoints/openresponses/sync_test.go diff --git a/core/http/endpoints/openresponses/responses.go b/core/http/endpoints/openresponses/responses.go index f8d741508b35..d86780fbd2c4 100644 --- a/core/http/endpoints/openresponses/responses.go +++ b/core/http/endpoints/openresponses/responses.go @@ -3047,11 +3047,26 @@ func GetResponseEndpoint() func(c echo.Context) error { // handleStreamResume handles resuming a streaming response from a specific sequence number func handleStreamResume(c echo.Context, store *ResponseStore, responseID string, stored *StoredResponse, startingAfter int) error { + // The resume buffer is process-local by design (see ErrResponseNotLocal), so + // a resume that the load balancer routed to a replica which is not + // generating this response cannot be served here. Say that plainly: a + // truthful error lets the caller retry against the right replica, whereas + // replaying an empty buffer would look like a stream that simply ended. + if stored.Remote { + return sendOpenResponsesError(c, 409, "invalid_request_error", + fmt.Sprintf("response %s is being generated on another replica (%s) and its stream cannot be resumed here; poll GET /v1/responses/%s instead, or route the resume with session affinity", responseID, stored.OwnerReplica, responseID), + "stream") + } + // Fetch buffered events before committing to an SSE response so an // offset-lost gap can be reported as a clean HTTP status rather than a // silently truncated event stream. events, err := store.GetEventsAfter(responseID, startingAfter) if err != nil { + if errors.Is(err, ErrResponseNotLocal) { + return sendOpenResponsesError(c, 409, "invalid_request_error", + fmt.Sprintf("response %s is being generated on another replica and its stream cannot be resumed here", responseID), "stream") + } if errors.Is(err, ErrOffsetLost) { return sendOpenResponsesError(c, 409, "invalid_request_error", fmt.Sprintf("starting_after=%d is older than the oldest retained event; the resume buffer evicted those events and the stream cannot be resumed from that point", startingAfter), "starting_after") } diff --git a/core/http/endpoints/openresponses/store.go b/core/http/endpoints/openresponses/store.go index ab52261e5c90..a4f1c8be7f58 100644 --- a/core/http/endpoints/openresponses/store.go +++ b/core/http/endpoints/openresponses/store.go @@ -9,6 +9,8 @@ import ( "time" "github.com/mudler/LocalAI/core/schema" + "github.com/mudler/LocalAI/core/services/messaging" + "github.com/mudler/LocalAI/core/services/syncstate" "github.com/mudler/xlog" ) @@ -49,6 +51,16 @@ type ResponseStore struct { // them. A value <= 0 disables that particular cap. maxStreamEvents int maxStreamBytes int + + // Cross-replica replication. All zero unless EnableDistributed was called + // (see sync.go), which is how a standalone deployment keeps exactly the + // previous process-local behaviour. Guarded by mu. + synced *syncstate.SyncedMap[string, *syncedResponse] + nats messaging.MessagingClient + cancelSub messaging.Subscription + replicaID string + lifeCtx context.Context + lifeCancel context.CancelFunc } // StreamedEvent represents a buffered SSE event for streaming resume @@ -80,6 +92,15 @@ type StoredResponse struct { EventsChan chan struct{} // Signals new events for live subscribers mu sync.RWMutex // Protect concurrent access to this response + // Remote marks a read-only view materialised from another replica's + // replicated metadata rather than from this process's own map. Such a view + // has no CancelFunc, no EventsChan and no resume buffer, so every path that + // needs one of those must check it and either delegate over the bus + // (cancel) or refuse with ErrResponseNotLocal (stream resume). + // OwnerReplica names the replica that does hold them. + Remote bool + OwnerReplica string + // streamBytes tracks the total serialized size of the events currently // retained in StreamEvents, used to enforce the byte cap. droppedThrough // is the highest sequence number evicted from the front of the buffer @@ -144,7 +165,6 @@ func NewResponseStore(ttl time.Duration) *ResponseStore { // Store stores a response with its request and items func (s *ResponseStore) Store(responseID string, request *schema.OpenResponsesRequest, response *schema.ORResponseResource) { s.mu.Lock() - defer s.mu.Unlock() // Build item index for quick lookup items := make(map[string]*schema.ORItemField) @@ -171,26 +191,40 @@ func (s *ResponseStore) Store(responseID string, request *schema.OpenResponsesRe } s.responses[responseID] = stored + s.mu.Unlock() + + // Replicate outside the lock: the broadcast can be delivered synchronously + // and a subscriber re-enters the store. + s.mirror(responseID, stored) xlog.Debug("Stored Open Responses response", "response_id", responseID, "items_count", len(items)) } -// Get retrieves a stored response by ID +// Get retrieves a stored response by ID. +// +// In distributed mode a miss in this process's map is not proof the response +// does not exist: a round-robin load balancer routes polls and +// previous_response_id lookups to any replica, not the one that created it. So +// a local miss falls back to the replicated metadata and returns a read-only +// remote view (issue #10993). Only a miss in both is a real "not found". func (s *ResponseStore) Get(responseID string) (*StoredResponse, error) { s.mu.RLock() - defer s.mu.RUnlock() - stored, exists := s.responses[responseID] - if !exists { - return nil, fmt.Errorf("response not found: %s", responseID) + s.mu.RUnlock() + + if exists { + // Check expiration + if stored.ExpiresAt != nil && time.Now().After(*stored.ExpiresAt) { + // Expired, but we'll return it anyway and let caller handle cleanup + return nil, fmt.Errorf("response expired: %s", responseID) + } + return stored, nil } - // Check expiration - if stored.ExpiresAt != nil && time.Now().After(*stored.ExpiresAt) { - // Expired, but we'll return it anyway and let caller handle cleanup - return nil, fmt.Errorf("response expired: %s", responseID) + if remote, ok := s.remoteGet(responseID); ok { + return remote, nil } - return stored, nil + return nil, fmt.Errorf("response not found: %s", responseID) } // GetItem retrieves a specific item from a stored response @@ -211,10 +245,41 @@ func (s *ResponseStore) GetItem(responseID, itemID string) (*schema.ORItemField, // FindItem searches for an item across all stored responses // Returns the item and the response ID it was found in func (s *ResponseStore) FindItem(itemID string) (*schema.ORItemField, string, error) { + now := time.Now() + + if item, responseID, found := s.findItemLocal(itemID, now); found { + return item, responseID, nil + } + + // An item referenced by ID can belong to a response created on any replica, + // so the local sweep alone reproduces the same coin-flip lookup #10993 + // describes for GET /v1/responses/{id}. + if m := s.syncMap(); m != nil { + for responseID, v := range m.Snapshot() { + if v == nil || v.Response == nil { + continue + } + if v.ExpiresAt != nil && now.After(*v.ExpiresAt) { + continue + } + for i := range v.Response.Output { + if v.Response.Output[i].ID == itemID { + return &v.Response.Output[i], responseID, nil + } + } + } + } + + return nil, "", fmt.Errorf("item not found in any stored response: %s", itemID) +} + +// findItemLocal sweeps only this process's own responses. Split out so the +// replicated sweep in FindItem runs with s.mu released - the SyncedMap is a +// separate component with its own lock and must not be entered under ours. +func (s *ResponseStore) findItemLocal(itemID string, now time.Time) (*schema.ORItemField, string, bool) { s.mu.RLock() defer s.mu.RUnlock() - now := time.Now() for responseID, stored := range s.responses { // Skip expired responses if stored.ExpiresAt != nil && now.After(*stored.ExpiresAt) { @@ -222,18 +287,19 @@ func (s *ResponseStore) FindItem(itemID string) (*schema.ORItemField, string, er } if item, exists := stored.Items[itemID]; exists { - return item, responseID, nil + return item, responseID, true } } - - return nil, "", fmt.Errorf("item not found in any stored response: %s", itemID) + return nil, "", false } // Delete removes a response from storage func (s *ResponseStore) Delete(responseID string) { s.mu.Lock() - defer s.mu.Unlock() delete(s.responses, responseID) + s.mu.Unlock() + + s.unmirror(responseID) xlog.Debug("Deleted Open Responses response", "response_id", responseID) } @@ -244,22 +310,33 @@ func (s *ResponseStore) Cleanup() int { } s.mu.Lock() - defer s.mu.Unlock() - now := time.Now() - count := 0 + expired := []string{} for id, stored := range s.responses { if stored.ExpiresAt != nil && now.After(*stored.ExpiresAt) { delete(s.responses, id) - count++ + expired = append(expired, id) + } + } + s.mu.Unlock() + + // Reap replicated entries too, including those whose owner never got to + // expire them because it was scaled down mid-flight. This covers the + // locally-expired IDs as well, since they were mirrored with the same + // ExpiresAt. Any replica may do this; the delete broadcast is idempotent. + if m := s.syncMap(); m != nil { + for id, v := range m.Snapshot() { + if v != nil && v.ExpiresAt != nil && now.After(*v.ExpiresAt) { + s.unmirror(id) + } } } - if count > 0 { - xlog.Debug("Cleaned up expired Open Responses", "count", count) + if len(expired) > 0 { + xlog.Debug("Cleaned up expired Open Responses", "count", len(expired)) } - return count + return len(expired) } // cleanupLoop runs periodic cleanup of expired responses @@ -292,7 +369,6 @@ func (s *ResponseStore) Count() int { // StoreBackground stores a background response with cancel function and optional streaming support func (s *ResponseStore) StoreBackground(responseID string, request *schema.OpenResponsesRequest, response *schema.ORResponseResource, cancelFunc context.CancelFunc, streamEnabled bool) { s.mu.Lock() - defer s.mu.Unlock() // Build item index for quick lookup items := make(map[string]*schema.ORItemField) @@ -324,6 +400,11 @@ func (s *ResponseStore) StoreBackground(responseID string, request *schema.OpenR } s.responses[responseID] = stored + s.mu.Unlock() + + // Only the metadata crosses the bus. CancelFunc and the resume buffer stay + // here, which is what makes this replica the owner for cancel and resume. + s.mirror(responseID, stored) xlog.Debug("Stored background Open Responses response", "response_id", responseID, "stream_enabled", streamEnabled) } @@ -338,10 +419,13 @@ func (s *ResponseStore) UpdateStatus(responseID string, status string, completed } stored.mu.Lock() - defer stored.mu.Unlock() - stored.Response.Status = status stored.Response.CompletedAt = completedAt + stored.mu.Unlock() + + // Peers poll this response too, so every status transition has to be + // republished or their view stays stuck at "queued" forever. + s.mirror(responseID, stored) xlog.Debug("Updated response status", "response_id", responseID, "status", status) return nil @@ -358,7 +442,6 @@ func (s *ResponseStore) UpdateResponse(responseID string, response *schema.ORRes } stored.mu.Lock() - defer stored.mu.Unlock() // Rebuild item index items := make(map[string]*schema.ORItemField) @@ -371,6 +454,10 @@ func (s *ResponseStore) UpdateResponse(responseID string, response *schema.ORRes stored.Response = response stored.Items = items + stored.mu.Unlock() + + // The final output is what a peer's poll must return, so replicate it. + s.mirror(responseID, stored) xlog.Debug("Updated response", "response_id", responseID, "status", response.Status, "items_count", len(items)) return nil @@ -436,6 +523,13 @@ func (s *ResponseStore) GetEventsAfter(responseID string, startingAfter int) ([] s.mu.RUnlock() if !exists { + // The response may be alive on a peer, whose resume buffer is not + // replicated. Say so explicitly instead of reporting "not found" (which + // invites the client to give up on a live stream) or an empty slice + // (which looks like a finished one). + if _, remote := s.remoteGet(responseID); remote { + return nil, ErrResponseNotLocal + } return nil, fmt.Errorf("response not found: %s", responseID) } @@ -460,16 +554,41 @@ func (s *ResponseStore) GetEventsAfter(responseID string, startingAfter int) ([] return result, nil } -// Cancel cancels a background response if it's still in progress +// Cancel cancels a background response if it's still in progress. +// +// The context.CancelFunc that actually stops generation is a function pointer +// and only exists in the process that created the response. When the cancel +// lands on any other replica - which a round-robin load balancer makes roughly +// as likely as landing on the right one - it is delegated to the owner over the +// bus instead of being reported as a 404 while generation keeps running +// (issue #10993). func (s *ResponseStore) Cancel(responseID string) (*schema.ORResponseResource, error) { s.mu.RLock() stored, exists := s.responses[responseID] s.mu.RUnlock() - if !exists { - return nil, fmt.Errorf("response not found: %s", responseID) + if exists { + response, err := s.cancelLocal(responseID, stored) + if err != nil { + return nil, err + } + s.mirror(responseID, stored) + return response, nil + } + + if m := s.syncMap(); m != nil { + if v, ok := m.Get(responseID); ok && v != nil { + return s.delegateCancel(v) + } } + return nil, fmt.Errorf("response not found: %s", responseID) +} + +// cancelLocal cancels a response this process owns. Shared by the direct HTTP +// path and by the delegated path a peer triggers over the bus, so both go +// through exactly the same terminal-state and CancelFunc handling. +func (s *ResponseStore) cancelLocal(responseID string, stored *StoredResponse) (*schema.ORResponseResource, error) { stored.mu.Lock() defer stored.mu.Unlock() @@ -502,6 +621,11 @@ func (s *ResponseStore) GetEventsChan(responseID string) (chan struct{}, error) s.mu.RUnlock() if !exists { + // A live subscriber channel cannot be handed across processes; see + // ErrResponseNotLocal. + if _, remote := s.remoteGet(responseID); remote { + return nil, ErrResponseNotLocal + } return nil, fmt.Errorf("response not found: %s", responseID) } @@ -515,6 +639,11 @@ func (s *ResponseStore) IsStreamEnabled(responseID string) (bool, error) { s.mu.RUnlock() if !exists { + // Stream-enabled is replicated metadata, so a peer can answer this one + // even though it cannot serve the buffer itself. + if remote, ok := s.remoteGet(responseID); ok { + return remote.StreamEnabled, nil + } return false, fmt.Errorf("response not found: %s", responseID) } @@ -541,6 +670,10 @@ func (s *ResponseStore) SetOwner(responseID, owner string) { } stored.Owner = owner + + // Owner is part of the replicated metadata: without it a peer would treat + // the response as ownerless and skip the access check in accessAllowed. + s.mirror(responseID, stored) } // accessAllowed reports whether a caller identified by callerID may read or diff --git a/core/http/endpoints/openresponses/sync.go b/core/http/endpoints/openresponses/sync.go new file mode 100644 index 000000000000..bd15b39cd8b6 --- /dev/null +++ b/core/http/endpoints/openresponses/sync.go @@ -0,0 +1,344 @@ +package openresponses + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/mudler/LocalAI/core/schema" + "github.com/mudler/LocalAI/core/services/messaging" + "github.com/mudler/LocalAI/core/services/syncstate" + "github.com/mudler/xlog" +) + +// syncStateName is the syncstate namespace for replicated response metadata. +// It becomes the NATS subject "state.responses.metadata.delta". +const syncStateName = "responses.metadata" + +// ErrResponseNotLocal is returned by the stream-resume accessors when the +// response exists in the cluster but its resume buffer lives in another +// replica's process memory. +// +// The buffer is a byte log plus a live notification channel: neither can be +// replicated without shipping every generated token over the message bus, so a +// resume request that lands on the wrong replica cannot be served. Returning a +// distinct error lets the HTTP layer say exactly that instead of handing the +// client an empty event list that looks like a completed stream. It is +// deliberately NOT ErrOffsetLost: that means "the buffer evicted the events you +// asked for" (a retention decision on the owning replica) and a client may +// reasonably give up on the stream. This one means "ask the replica that owns +// it", which is a routing problem, and the two must stay distinguishable. +var ErrResponseNotLocal = errors.New("response is owned by another replica: its stream resume buffer is not available here") + +// syncedResponse is the replicable projection of a StoredResponse. +// +// Only what a peer can act on is carried: the request (so previous_response_id +// chaining can replay it), the response resource (so polling returns real +// state), the expiry, and the owning replica. Everything process-local is +// excluded by construction - the CancelFunc is a function pointer, EventsChan is +// a live channel, and StreamEvents is an unbounded byte log. Cancel is therefore +// delegated to the owner over the bus rather than replicated (see +// delegateCancel), and stream resume is refused off-owner (ErrResponseNotLocal). +type syncedResponse struct { + ID string `json:"id"` + OwnerReplica string `json:"owner_replica"` + Owner string `json:"owner,omitempty"` + Request *schema.OpenResponsesRequest `json:"request,omitempty"` + Response *schema.ORResponseResource `json:"response,omitempty"` + StoredAt time.Time `json:"stored_at"` + ExpiresAt *time.Time `json:"expires_at,omitempty"` + StreamEnabled bool `json:"stream_enabled,omitempty"` + IsBackground bool `json:"is_background,omitempty"` +} + +// responseCancelEvent is the wire envelope broadcast on SubjectResponseCancel. +// Origin lets the publisher skip its own echo: it has already cancelled locally +// (or knows it holds nothing to cancel) before publishing. +type responseCancelEvent struct { + ResponseID string `json:"response_id"` + Origin string `json:"origin"` +} + +// EnableDistributed turns on cross-replica replication for this store. +// +// It is called once during route registration when the application is in +// distributed mode, before any request is served. Standalone deployments never +// call it and keep the store exactly as it was: a process-local map with no +// broadcast, no subscription and no behavioural change. +// +// Two independent legs are wired: +// +// - a syncstate.SyncedMap carrying response metadata, so a GET or a +// previous_response_id lookup that a round-robin load balancer sends to a +// replica which did not create the response still resolves; +// - a wildcard subscription on the response-cancel subject, so a cancel that +// lands on the wrong replica still reaches the context.CancelFunc. +// +// The SyncedMap has no durable Store: responses are ephemeral, TTL-bounded +// state that today does not survive a process restart either, so peers converge +// through deltas alone. A replica that joins later does not learn about +// responses created before it started; that is the same visibility a client had +// before this change and strictly better than the 404 it got from every peer. +func (s *ResponseStore) EnableDistributed(ctx context.Context, nats messaging.MessagingClient, replicaID string) error { + if nats == nil { + return nil + } + + // The subscription callbacks and the delegated-cancel publisher outlive the + // caller's ctx (which is startup-scoped), so hold a context that only Close + // cancels. + lifeCtx, lifeCancel := context.WithCancel(context.Background()) //#nosec G118 -- cancelled in Close() + + synced := syncstate.New(syncstate.Config[string, *syncedResponse]{ + Name: syncStateName, + Key: func(v *syncedResponse) string { return v.ID }, + Nats: nats, + }) + if err := synced.Start(ctx); err != nil { + lifeCancel() + return fmt.Errorf("starting response metadata sync: %w", err) + } + + // Publish the wiring before subscribing: the subscription handler re-enters + // the store, so everything it reads has to be in place first. + s.mu.Lock() + s.replicaID = replicaID + s.nats = nats + s.lifeCtx, s.lifeCancel = lifeCtx, lifeCancel + s.synced = synced + s.mu.Unlock() + + sub, err := messaging.SubscribeJSON(nats, messaging.SubjectResponseCancelWildcard, s.applyRemoteCancel) + if err != nil { + if cerr := s.Close(); cerr != nil { + xlog.Warn("failed to tear down response metadata sync after subscribe error", "error", cerr) + } + return fmt.Errorf("subscribing to response cancel wildcard: %w", err) + } + + s.mu.Lock() + s.cancelSub = sub + s.mu.Unlock() + + xlog.Info("Open Responses store replicating across replicas", "replica_id", replicaID) + return nil +} + +// Close tears down the distributed wiring. It is idempotent so a test (or a +// double shutdown) can call it more than once, and is a no-op for a standalone +// store. +func (s *ResponseStore) Close() error { + s.mu.Lock() + sub := s.cancelSub + s.cancelSub = nil + synced := s.synced + s.synced = nil + cancel := s.lifeCancel + s.lifeCancel = nil + s.mu.Unlock() + + if cancel != nil { + cancel() + } + if sub != nil { + if err := sub.Unsubscribe(); err != nil { + return err + } + } + if synced != nil { + return synced.Close() + } + return nil +} + +// syncMap returns the replicated metadata map, or nil in standalone mode. +func (s *ResponseStore) syncMap() *syncstate.SyncedMap[string, *syncedResponse] { + s.mu.RLock() + defer s.mu.RUnlock() + return s.synced +} + +// distributed returns the replication handles as a consistent snapshot. Every +// path that broadcasts reads them through here so a concurrent Close cannot be +// observed half-applied. A nil map means standalone mode. +func (s *ResponseStore) distributed() (*syncstate.SyncedMap[string, *syncedResponse], context.Context, messaging.MessagingClient, string) { + s.mu.RLock() + defer s.mu.RUnlock() + ctx := s.lifeCtx + if ctx == nil { + ctx = context.Background() + } + return s.synced, ctx, s.nats, s.replicaID +} + +// replicaIdentity returns this process's replica ID (empty in standalone mode). +func (s *ResponseStore) replicaIdentity() string { + s.mu.RLock() + defer s.mu.RUnlock() + return s.replicaID +} + +// mirror publishes the current metadata of a locally-owned response to peers. +// +// It must be called with s.mu released: the broadcast is delivered synchronously +// by the in-memory bus used in tests, and a subscriber callback re-enters the +// store to look up the response. +func (s *ResponseStore) mirror(responseID string, stored *StoredResponse) { + m, ctx, _, replicaID := s.distributed() + if m == nil { + return + } + + stored.mu.RLock() + v := &syncedResponse{ + ID: responseID, + OwnerReplica: replicaID, + Owner: stored.Owner, + Request: stored.Request, + Response: stored.Response, + StoredAt: stored.StoredAt, + ExpiresAt: stored.ExpiresAt, + StreamEnabled: stored.StreamEnabled, + IsBackground: stored.IsBackground, + } + stored.mu.RUnlock() + + if err := m.Set(ctx, v); err != nil { + xlog.Warn("failed to replicate Open Responses metadata", "response_id", responseID, "error", err) + } +} + +// unmirror removes a response from the replicated metadata. +func (s *ResponseStore) unmirror(responseID string) { + m, ctx, _, _ := s.distributed() + if m == nil { + return + } + if err := m.Delete(ctx, responseID); err != nil { + xlog.Warn("failed to remove replicated Open Responses metadata", "response_id", responseID, "error", err) + } +} + +// remoteGet resolves a response from the replicated metadata and materialises a +// read-only StoredResponse view of it. +// +// The view is rebuilt on every call rather than inserted into s.responses: it +// carries no CancelFunc and no resume buffer, so caching it locally would make +// the store unable to tell "mine" from "a peer's" and would silently break the +// stream and cancel paths. Remote is the flag every process-local accessor keys +// off. +func (s *ResponseStore) remoteGet(responseID string) (*StoredResponse, bool) { + m := s.syncMap() + if m == nil { + return nil, false + } + v, ok := m.Get(responseID) + if !ok || v == nil || v.Response == nil { + return nil, false + } + if v.ExpiresAt != nil && time.Now().After(*v.ExpiresAt) { + return nil, false + } + + // Rebuild the item index on this side so GetItem/FindItem behave the same + // as they do on the owner; the index is derived data and is not shipped. + items := make(map[string]*schema.ORItemField) + for i := range v.Response.Output { + item := &v.Response.Output[i] + if item.ID != "" { + items[item.ID] = item + } + } + + return &StoredResponse{ + Request: v.Request, + Response: v.Response, + Items: items, + StoredAt: v.StoredAt, + ExpiresAt: v.ExpiresAt, + Owner: v.Owner, + StreamEnabled: v.StreamEnabled, + IsBackground: v.IsBackground, + Remote: true, + OwnerReplica: v.OwnerReplica, + droppedThrough: -1, + }, true +} + +// delegateCancel handles a cancel for a response this replica does not own. +// +// The CancelFunc lives in the owner's process, so the request is broadcast and +// the owner applies it locally (applyRemoteCancel). The broadcast is +// fire-and-forget: a request/reply would block this HTTP handler for a timeout +// whenever the owner has crashed or been scaled down, which is exactly the case +// where the caller most needs a prompt answer. +// +// The replicated status is moved to cancelled here rather than waiting for the +// owner to confirm. If the owner is alive it will cancel and converge on the +// same value; if it is gone its generation died with its process, so cancelled +// is the truthful terminal state either way. +func (s *ResponseStore) delegateCancel(v *syncedResponse) (*schema.ORResponseResource, error) { + if v.Response == nil { + return nil, fmt.Errorf("response not found: %s", v.ID) + } + + status := v.Response.Status + if status == schema.ORStatusCompleted || status == schema.ORStatusFailed || + status == schema.ORStatusIncomplete || status == schema.ORStatusCancelled { + xlog.Debug("Peer-owned response already in terminal state", "response_id", v.ID, "status", status) + return v.Response, nil + } + + m, ctx, nats, replicaID := s.distributed() + if nats != nil { + if err := nats.Publish(messaging.SubjectResponseCancel(v.ID), + responseCancelEvent{ResponseID: v.ID, Origin: replicaID}); err != nil { + xlog.Warn("failed to broadcast Open Responses cancel", "response_id", v.ID, "error", err) + } + } + + // Copy before mutating: the value in the SyncedMap is shared with any peer + // delta already applied, and the response resource is handed to the client. + updated := *v + response := *v.Response + response.Status = schema.ORStatusCancelled + now := time.Now().Unix() + response.CompletedAt = &now + updated.Response = &response + + if m != nil { + if err := m.Set(ctx, &updated); err != nil { + xlog.Warn("failed to replicate cancelled Open Responses status", "response_id", v.ID, "error", err) + } + } + + xlog.Debug("Delegated Open Responses cancel to owning replica", "response_id", v.ID, "owner_replica", v.OwnerReplica) + return &response, nil +} + +// applyRemoteCancel runs a peer's delegated cancel against local state. +// +// Only the replica that actually holds the response in s.responses has anything +// to do; every other subscriber drops the event. It deliberately does not +// re-broadcast and does not re-write the replicated metadata - the delegating +// replica already published the cancelled status - which is the same echo-loop +// guard syncstate applies on its own apply path. +func (s *ResponseStore) applyRemoteCancel(evt responseCancelEvent) { + if evt.ResponseID == "" || evt.Origin == s.replicaIdentity() { + return + } + + s.mu.RLock() + stored, exists := s.responses[evt.ResponseID] + s.mu.RUnlock() + if !exists { + return + } + + if _, err := s.cancelLocal(evt.ResponseID, stored); err != nil { + xlog.Warn("failed to apply delegated Open Responses cancel", "response_id", evt.ResponseID, "error", err) + return + } + xlog.Debug("Applied delegated Open Responses cancel", "response_id", evt.ResponseID, "origin", evt.Origin) +} diff --git a/core/http/endpoints/openresponses/sync_test.go b/core/http/endpoints/openresponses/sync_test.go new file mode 100644 index 000000000000..ef0a4cad1a6c --- /dev/null +++ b/core/http/endpoints/openresponses/sync_test.go @@ -0,0 +1,208 @@ +package openresponses + +import ( + "context" + "errors" + "time" + + "github.com/mudler/LocalAI/core/schema" + "github.com/mudler/LocalAI/core/services/testutil" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +// These specs model the two-replica topology from issue #10993: two independent +// ResponseStore instances (one per frontend process) sharing a single message +// bus. Anything a client can observe through the HTTP API after a round-robin +// load balancer sends it to the "wrong" replica must be asserted here. +var _ = Describe("ResponseStore cross-replica", func() { + var ( + bus *testutil.FakeBus + replicaA *ResponseStore + replicaB *ResponseStore + ctx context.Context + ) + + BeforeEach(func() { + ctx = context.Background() + bus = testutil.NewFakeBus() + + replicaA = NewResponseStore(0) + replicaB = NewResponseStore(0) + + Expect(replicaA.EnableDistributed(ctx, bus, "replica-a")).To(Succeed()) + Expect(replicaB.EnableDistributed(ctx, bus, "replica-b")).To(Succeed()) + }) + + AfterEach(func() { + Expect(replicaA.Close()).To(Succeed()) + Expect(replicaB.Close()).To(Succeed()) + }) + + newResponse := func(id, status string) *schema.ORResponseResource { + return &schema.ORResponseResource{ + ID: id, + Object: "response", + CreatedAt: time.Now().Unix(), + Status: status, + Model: "test-model", + Output: []schema.ORItemField{ + {Type: "message", ID: "msg_" + id, Role: "assistant"}, + }, + } + } + + Describe("polling and previous_response_id chaining", func() { + It("makes a response created on one replica readable on its peer", func() { + const id = "resp_cross_replica" + request := &schema.OpenResponsesRequest{Model: "test-model", Input: "Hello"} + + replicaA.Store(id, request, newResponse(id, schema.ORStatusCompleted)) + + // The peer never saw the POST that created this response; without + // replication this is the HTTP 404 reported in #10993. + stored, err := replicaB.Get(id) + Expect(err).ToNot(HaveOccurred()) + Expect(stored).ToNot(BeNil()) + Expect(stored.Response.Status).To(Equal(schema.ORStatusCompleted)) + // previous_response_id chaining replays the original request, so the + // request body has to survive the hop, not just the response. + Expect(stored.Request).ToNot(BeNil()) + Expect(stored.Request.Model).To(Equal("test-model")) + // Item lookup is rebuilt on the peer so GetItem/FindItem work there too. + Expect(stored.Items).To(HaveKey("msg_" + id)) + }) + + It("propagates status updates from the owner to the peer", func() { + const id = "resp_status_update" + replicaA.StoreBackground(id, &schema.OpenResponsesRequest{Model: "test-model"}, + newResponse(id, schema.ORStatusQueued), func() {}, false) + + completedAt := time.Now().Unix() + Expect(replicaA.UpdateStatus(id, schema.ORStatusCompleted, &completedAt)).To(Succeed()) + + stored, err := replicaB.Get(id) + Expect(err).ToNot(HaveOccurred()) + Expect(stored.Response.Status).To(Equal(schema.ORStatusCompleted)) + }) + + It("removes a deleted response from the peer as well", func() { + const id = "resp_deleted" + replicaA.Store(id, &schema.OpenResponsesRequest{Model: "test-model"}, newResponse(id, schema.ORStatusCompleted)) + Expect(replicaB.Get(id)).ToNot(BeNil()) + + replicaA.Delete(id) + + _, err := replicaB.Get(id) + Expect(err).To(HaveOccurred()) + }) + + It("still reports a genuinely unknown response as not found", func() { + _, err := replicaB.Get("resp_never_existed") + Expect(err).To(HaveOccurred()) + }) + }) + + Describe("cancel delegation", func() { + It("reaches the CancelFunc held by the owning replica", func() { + const id = "resp_cancel_delegated" + cancelled := make(chan struct{}) + replicaA.StoreBackground(id, &schema.OpenResponsesRequest{Model: "test-model"}, + newResponse(id, schema.ORStatusInProgress), func() { close(cancelled) }, false) + + // Cancel lands on the replica that does NOT hold the CancelFunc. Today + // this 404s and generation keeps burning GPU on replica A. + resp, err := replicaB.Cancel(id) + Expect(err).ToNot(HaveOccurred()) + Expect(resp.Status).To(Equal(schema.ORStatusCancelled)) + + Eventually(cancelled).Should(BeClosed()) + + // The owner's own view must converge too, so a later poll on A does + // not report the response as still in progress. + ownerView, err := replicaA.Get(id) + Expect(err).ToNot(HaveOccurred()) + Expect(ownerView.Response.Status).To(Equal(schema.ORStatusCancelled)) + }) + + It("does not hang when the owning replica is gone", func() { + const id = "resp_dead_owner" + replicaA.StoreBackground(id, &schema.OpenResponsesRequest{Model: "test-model"}, + newResponse(id, schema.ORStatusInProgress), func() {}, false) + + // Simulate the owner crashing / being scaled down: it stops consuming + // the bus, so nothing will ever answer a delegated cancel. The peer + // still holds the replicated metadata and must resolve on its own. + Expect(replicaA.Close()).To(Succeed()) + + done := make(chan struct{}) + go func() { + defer GinkgoRecover() + defer close(done) + resp, err := replicaB.Cancel(id) + Expect(err).ToNot(HaveOccurred()) + Expect(resp.Status).To(Equal(schema.ORStatusCancelled)) + }() + Eventually(done, 5*time.Second).Should(BeClosed()) + }) + + It("is idempotent for a response already in a terminal state", func() { + const id = "resp_cancel_terminal" + replicaA.Store(id, &schema.OpenResponsesRequest{Model: "test-model"}, newResponse(id, schema.ORStatusCompleted)) + + resp, err := replicaB.Cancel(id) + Expect(err).ToNot(HaveOccurred()) + Expect(resp.Status).To(Equal(schema.ORStatusCompleted)) + }) + }) + + Describe("stream resume", func() { + It("reports a distinct error instead of a truncated stream when the buffer lives on a peer", func() { + const id = "resp_stream_remote" + replicaA.StoreBackground(id, &schema.OpenResponsesRequest{Model: "test-model"}, + newResponse(id, schema.ORStatusInProgress), func() {}, true) + Expect(replicaA.AppendEvent(id, &schema.ORStreamEvent{SequenceNumber: 1, Type: "response.created"})).To(Succeed()) + + // The resume buffer is a byte buffer on replica A and is deliberately + // not replicated; asking B for it must be an explicit error, never an + // empty (silently truncated) event list. + events, err := replicaB.GetEventsAfter(id, 0) + Expect(events).To(BeEmpty()) + Expect(errors.Is(err, ErrResponseNotLocal)).To(BeTrue()) + + // ErrOffsetLost means "the buffer evicted your events"; a peer lookup + // is a different condition and must not be conflated with it. + Expect(errors.Is(err, ErrOffsetLost)).To(BeFalse()) + + _, err = replicaB.GetEventsChan(id) + Expect(errors.Is(err, ErrResponseNotLocal)).To(BeTrue()) + }) + + It("keeps serving the resume buffer on the owning replica", func() { + const id = "resp_stream_local" + replicaA.StoreBackground(id, &schema.OpenResponsesRequest{Model: "test-model"}, + newResponse(id, schema.ORStatusInProgress), func() {}, true) + Expect(replicaA.AppendEvent(id, &schema.ORStreamEvent{SequenceNumber: 1, Type: "response.created"})).To(Succeed()) + + events, err := replicaA.GetEventsAfter(id, 0) + Expect(err).ToNot(HaveOccurred()) + Expect(events).To(HaveLen(1)) + }) + }) + + Describe("standalone mode", func() { + It("keeps a store with no messaging client purely local", func() { + standalone := NewResponseStore(0) + const id = "resp_standalone" + standalone.Store(id, &schema.OpenResponsesRequest{Model: "test-model"}, newResponse(id, schema.ORStatusCompleted)) + + stored, err := standalone.Get(id) + Expect(err).ToNot(HaveOccurred()) + Expect(stored).ToNot(BeNil()) + + // Nothing was broadcast, so peers on a shared bus stay unaware. + _, err = replicaB.Get(id) + Expect(err).To(HaveOccurred()) + }) + }) +}) diff --git a/core/http/routes/openresponses.go b/core/http/routes/openresponses.go index a5932fb0f94b..8aff6ccf13a1 100644 --- a/core/http/routes/openresponses.go +++ b/core/http/routes/openresponses.go @@ -9,6 +9,7 @@ import ( "github.com/mudler/LocalAI/core/http/endpoints/openresponses" "github.com/mudler/LocalAI/core/http/middleware" "github.com/mudler/LocalAI/core/schema" + "github.com/mudler/xlog" ) func RegisterOpenResponsesRoutes(app *echo.Echo, @@ -19,6 +20,16 @@ func RegisterOpenResponsesRoutes(app *echo.Echo, var natsClient mcpTools.MCPNATSClient if d := application.Distributed(); d != nil { natsClient = d.Nats + + // Replicate response metadata across frontend replicas and subscribe to + // delegated cancels. Without this a GET, a previous_response_id lookup or + // a cancel that the load balancer sends to a replica other than the + // creator 404s, and the cancel never reaches the CancelFunc (#10993). + // Standalone deployments skip this entirely and stay process-local. + if err := openresponses.GetGlobalStore().EnableDistributed( + application.ApplicationConfig().Context, d.Nats, application.InstanceID()); err != nil { + xlog.Error("Failed to enable cross-replica Open Responses store", "error", err) + } } // Open Responses API endpoint diff --git a/core/services/messaging/subjects.go b/core/services/messaging/subjects.go index dd05fe2ede64..11e3bfba97e0 100644 --- a/core/services/messaging/subjects.go +++ b/core/services/messaging/subjects.go @@ -96,6 +96,7 @@ const ( subjectAgentCancelPrefix = "agent." subjectFineTuneCancelPrefix = "finetune." subjectGalleryCancelPrefix = "gallery." + subjectResponseCancelPrefix = "responses." ) // Wildcard subjects for NATS subscriptions that match all IDs. @@ -106,6 +107,7 @@ const ( SubjectAgentCancelWildcard = "agent.*.cancel" SubjectGalleryCancelWildcard = "gallery.*.cancel" SubjectGalleryProgressWildcard = "gallery.*.progress" + SubjectResponseCancelWildcard = "responses.*.cancel" ) // SubjectJobCancel returns the NATS subject to cancel a running job. @@ -128,6 +130,17 @@ func SubjectGalleryCancel(opID string) string { return subjectGalleryCancelPrefix + sanitizeSubjectToken(opID) + ".cancel" } +// SubjectResponseCancel returns the NATS subject used to cancel an in-flight +// Open Responses generation. Only the replica that created the response holds +// its context.CancelFunc, so a cancel that lands on any other replica is +// broadcast here and applied by whichever replica actually owns the function. +// Broadcast rather than request/reply on purpose: if the owner crashed or was +// scaled down, nobody answers and the caller must not block waiting for a +// reply that will never come. +func SubjectResponseCancel(responseID string) string { + return subjectResponseCancelPrefix + sanitizeSubjectToken(responseID) + ".cancel" +} + // Node Backend Lifecycle (Pub/Sub — targeted to specific nodes) // // These subjects control the backend *process* lifecycle on a serve-backend node, diff --git a/docs/content/features/text-generation.md b/docs/content/features/text-generation.md index a245a9f618a3..f7cc155a80f5 100644 --- a/docs/content/features/text-generation.md +++ b/docs/content/features/text-generation.md @@ -331,6 +331,22 @@ Cancel a background response that's still in progress: curl -X POST http://localhost:8080/v1/responses/resp_abc123/cancel ``` +#### Multiple Replicas (Distributed Mode) + +In distributed mode LocalAI replicates response metadata across frontend +replicas, so retrieval, `previous_response_id` chaining and cancellation work +regardless of which replica the load balancer picks: + +- `GET /v1/responses/{id}` returns the response from any replica. +- `POST /v1/responses/{id}/cancel` is delegated over NATS to the replica that is + actually generating, so generation really stops. If that replica is gone, the + response is reported as `cancelled` without blocking. +- **Streaming resume (`?stream=true`) is served only by the replica that created + the response.** The event buffer lives in that process's memory and is not + replicated. A resume request that reaches another replica returns HTTP 409 + naming the owning replica instead of silently returning a truncated stream. + Poll the response instead, or route resume requests with session affinity. + #### Tool Calling Open Responses API supports function calling with tools: