diff --git a/devshard/cmd/devshardctl/escrow_rotator.go b/devshard/cmd/devshardctl/escrow_rotator.go index 1300a419ab..e8d5c5e830 100644 --- a/devshard/cmd/devshardctl/escrow_rotator.go +++ b/devshard/cmd/devshardctl/escrow_rotator.go @@ -391,9 +391,43 @@ func (g *Gateway) settleDevshardOnChain(ctx context.Context, id string, req admi rt.active.Store(false) } g.mu.Unlock() + wasResident := ok if !ok { - log.Printf("devshard_settle_failed escrow=%s stage=runtime_lookup error=%q", id, "devshard is not active") - return nil, fmt.Errorf("devshard %s is not active", id) + // Non-resident devshard (inactive/settled): rehydrate a full runtime + // (with chain access) solely to build and broadcast this settlement, + // then release it. Concurrent settlement of the same id is guarded by + // the caller: auto/reconcile paths hold settlementInFlight[id] via + // scheduleAutoSettlement, and the chain rejects any duplicate settle + // broadcast, so no additional in-flight guard is taken here (taking + // one would deadlock the auto path, which already holds it). + cfg, known, cfgErr := g.lazyRuntimeConfig(id) + if cfgErr != nil { + log.Printf("devshard_settle_failed escrow=%s stage=lazy_config error=%q", id, cfgErr.Error()) + return nil, cfgErr + } + if !known { + log.Printf("devshard_settle_failed escrow=%s stage=runtime_lookup error=%q", id, "devshard is not active") + return nil, fmt.Errorf("devshard %s is not active", id) + } + g.mu.Lock() + settings := g.settings + g.mu.Unlock() + built, buildErr := gatewayRuntimeBuilder(cfg, settings.ChainREST, settings.DefaultModel, g.perf) + if buildErr != nil { + log.Printf("devshard_settle_failed escrow=%s stage=rehydrate error=%q", id, buildErr.Error()) + return nil, fmt.Errorf("rehydrate devshard %s for settlement: %w", id, buildErr) + } + built.active.Store(false) + rt = built + log.Printf("devshard_settle_rehydrated escrow=%s (transient, non-resident)", id) + defer func() { + // Flush a final snapshot: Finalize advances the nonce, so a later + // read-only rebuild of this settled escrow would otherwise replay + // the diff tail. retireClose captures the finalized state once. + if closeErr := rt.retireClose("settled-transient"); closeErr != nil { + log.Printf("devshard_settle_transient_close_error escrow=%s error=%v", id, closeErr) + } + }() } if err := g.store.SetDevshardActive(id, false); err != nil { log.Printf("devshard_settle_failed escrow=%s stage=persist_deactivate error=%q", id, err.Error()) @@ -445,5 +479,13 @@ func (g *Gateway) settleDevshardOnChain(ctx context.Context, id string, req admi return nil, err } log.Printf("devshard_settle_submitted escrow=%s tx_hash=%s settler=%s", id, result.TxHash, result.Settler) + g.clearSettlementPending(id) + // A settled escrow is terminal: drop the resident runtime so its memory + // (state machine, inference map, SQLite handles) is released now rather + // than lingering until the next restart. Transient runtimes are closed by + // their own deferred cleanup above. + if wasResident { + g.retireRuntime(id, "settled") + } return result, nil } diff --git a/devshard/cmd/devshardctl/gateway.go b/devshard/cmd/devshardctl/gateway.go index 61859b58dc..9db3ff7765 100644 --- a/devshard/cmd/devshardctl/gateway.go +++ b/devshard/cmd/devshardctl/gateway.go @@ -10,9 +10,11 @@ import ( "log" "math" "net/http" + "net/http/pprof" "net/url" "os" "path/filepath" + "runtime" "slices" "strconv" "strings" @@ -84,6 +86,18 @@ type devshardRuntime struct { activeRequests atomic.Int64 reservedTokens atomic.Int64 + // settlementPending marks an escrow that has been deactivated and must + // be settled once its in-flight requests drain. settlementReason is + // written before the flag and read after it in the lock-free drain hook; + // the atomic Store→Load pair supplies the happens-before. + settlementPending atomic.Bool + settlementReason string + + // retirePending marks a runtime whose retirement was deferred because a + // request was still in flight; + retirePending atomic.Bool + retireReason string + activeConfigured bool } @@ -97,6 +111,7 @@ type runtimeStatus struct { ProtocolVersion string `json:"protocol_version,omitempty"` ActiveRequests int64 `json:"active_requests"` ReservedTokens int64 `json:"reserved_tokens"` + SettlementPending bool `json:"settlement_pending,omitempty"` ChainPhase string `json:"chain_phase,omitempty"` ConfirmationPoCPhase string `json:"confirmation_poc_phase,omitempty"` RequestsBlocked bool `json:"requests_blocked"` @@ -207,6 +222,7 @@ func newRuntimeMux(proxy *Proxy) http.Handler { mux.HandleFunc("GET /v1/state", proxy.handleState) mux.HandleFunc("GET /v1/debug/pending", proxy.handleDebugPending) mux.HandleFunc("GET /v1/debug/state", proxy.handleDebugState) + mux.HandleFunc("GET /v1/debug/inferences", proxy.handleDebugInferences) mux.HandleFunc("GET /v1/debug/perf", proxy.handleDebugPerf) mux.HandleFunc("GET /v1/debug/pairwise", proxy.handleDebugPairwise) mux.HandleFunc("GET /v1/debug/signatures", proxy.handleDebugSignatures) @@ -305,6 +321,125 @@ func buildRuntime(cfg RuntimeConfig, chainREST, defaultModel string, perf *PerfT return rt, nil } +// buildReadOnlyRuntime rehydrates a transient, read-only runtime from local +// storage without contacting the chain. It is used to serve debug/state +// endpoints for devshards that are not resident in memory (inactive or +// settled escrows). The runtime has no host clients and no redundancy: it can +// answer read queries but must never route inferences. Callers own the +// returned runtime and must close() it once the response is served. +func buildReadOnlyRuntime(cfg RuntimeConfig, defaultModel string, perf *PerfTracker) (*devshardRuntime, error) { + keyHex := strings.TrimSpace(cfg.PrivateKeyHex) + if keyHex == "" && cfg.PrivateKeyEnv != "" { + keyHex = strings.TrimSpace(os.Getenv(cfg.PrivateKeyEnv)) + } + if keyHex == "" { + return nil, fmt.Errorf("runtime %s: %w", cfg.ID, errRuntimePrivateKeyMissing) + } + model := cfg.Model + if model == "" { + model = defaultModel + } + cfg.StoragePath = normalizeStorageDir(cfg.StoragePath) + pv, pvErr := types.ParseProtocolVersion(cfg.ProtocolVersion) + if pvErr != nil { + return nil, fmt.Errorf("runtime %s: %w", cfg.ID, pvErr) + } + if perf == nil { + perf = NewPerfTracker(nil) + } + session, sm, err := user.NewLocalSession(user.LocalSessionConfig{ + PrivateKeyHex: keyHex, + EscrowID: cfg.ID, + StoragePath: cfg.StoragePath, + ProtocolVersion: pv, + }) + if err != nil { + return nil, fmt.Errorf("runtime %s: rehydrate local session: %w", cfg.ID, err) + } + proxy := &Proxy{ + session: session, + sm: sm, + escrowID: cfg.ID, + model: model, + perf: perf, + } + rt := &devshardRuntime{ + id: cfg.ID, + model: model, + handler: newRuntimeMux(proxy), + proxy: proxy, + session: session, + participantKeys: session.ParticipantKeys(), + } + rt.active.Store(false) + rt.activeConfigured = true + return rt, nil +} + +// lazyRuntimeConfig resolves a non-resident devshard's runtime config from the +// registry store, applying the same finalization (storage path, default +// model) used at boot. It returns false when the devshard is unknown. +func (g *Gateway) lazyRuntimeConfig(id string) (RuntimeConfig, bool, error) { + if g.store == nil { + return RuntimeConfig{}, false, fmt.Errorf("gateway state store unavailable") + } + record, ok, err := g.store.GetDevshard(id) + if err != nil { + return RuntimeConfig{}, false, err + } + if !ok { + return RuntimeConfig{}, false, nil + } + g.mu.Lock() + defaultModel := g.settings.DefaultModel + baseStorageDir := g.baseStorageDir + g.mu.Unlock() + cfgs, err := finalizeRuntimeConfigs([]RuntimeConfig{record.RuntimeConfig}, defaultModel, baseStorageDir) + if err != nil { + return RuntimeConfig{}, false, err + } + return cfgs[0], true, nil +} + +// hydrateReadOnlyRuntime builds a transient read-only runtime for a +// non-resident devshard, serving from local storage only (no chain). Returns +// (nil, false, nil) when the devshard is unknown to the registry. +func (g *Gateway) hydrateReadOnlyRuntime(id string) (*devshardRuntime, bool, error) { + cfg, ok, err := g.lazyRuntimeConfig(id) + if err != nil || !ok { + return nil, ok, err + } + rt, err := buildReadOnlyRuntime(cfg, g.settings.DefaultModel, g.perf) + if err != nil { + return nil, true, err + } + return rt, true, nil +} + +// isReadOnlyDevshardPath reports whether an inner devshard path may be served +// by a transient read-only (no-chain, no-clients) runtime. Only idempotent +// GET reads qualify; anything that dispatches inferences or mutates state is +// excluded so it never runs against a client-less runtime. +func isReadOnlyDevshardPath(method, innerPath string) bool { + if method != http.MethodGet && method != http.MethodHead { + return false + } + switch innerPath { + case "/v1/status", + "/v1/state", + "/v1/finalize", + "/v1/models", + "/v1/debug/pending", + "/v1/debug/state", + "/v1/debug/inferences", + "/v1/debug/perf", + "/v1/debug/pairwise", + "/v1/debug/signatures": + return true + } + return strings.HasPrefix(innerPath, "/v1/requests/") +} + func newRESTBridgeForProtocol(chainREST string, pv types.ProtocolVersion) *bridge.RESTBridge { return bridge.NewRESTBridge(chainREST) } @@ -349,6 +484,21 @@ func (rt *devshardRuntime) close() error { return nil } +// retireClose flushes a final state snapshot at the current (now frozen) nonce +// so the escrow can later be rebuilt -- read-only for debug/state or fully on +// reactivation -- without replaying the diff tail accumulated since the last +// periodic snapshot, then closes the runtime. Use only on retire paths +// (deactivate, settle, rotation); plain close() is for transient read-only +// runtimes and build-failure cleanup, where no snapshot flush is wanted. +func (rt *devshardRuntime) retireClose(reason string) error { + if rt.session != nil { + if err := rt.session.FlushSnapshot(); err != nil { + log.Printf("runtime_retire_snapshot_error escrow=%s reason=%q error=%v", rt.id, reason, err) + } + } + return rt.close() +} + func (rt *devshardRuntime) acceptsNewInferences() (bool, string) { if rt == nil || !rt.active.Load() { return false, "inactive" @@ -406,18 +556,18 @@ func sessionPhaseLabel(phase types.SessionPhase) string { func (rt *devshardRuntime) snapshot() runtimeStatus { status := runtimeStatus{ - ID: rt.id, - Model: rt.model, - Active: rt.active.Load(), - ActiveRequests: rt.activeRequests.Load(), - ReservedTokens: rt.reservedTokens.Load(), + ID: rt.id, + Model: rt.model, + Active: rt.active.Load(), + ActiveRequests: rt.activeRequests.Load(), + ReservedTokens: rt.reservedTokens.Load(), + SettlementPending: rt.settlementPending.Load(), } if rt.proxy != nil && rt.proxy.sm != nil && rt.proxy.session != nil { phase := rt.proxy.sm.Phase() status.Phase = sessionPhaseLabel(phase) - st := rt.proxy.sm.SnapshotState() status.Nonce = rt.proxy.session.Nonce() - status.Balance = st.Balance + status.Balance = rt.proxy.sm.Balance() status.ProtocolVersion = string(rt.proxy.sm.ProtocolVersion()) } if rt.proxy != nil && rt.proxy.phaseGate != nil { @@ -482,7 +632,7 @@ func NewGateway(runtimes []*devshardRuntime, limiter *GatewayLimiter, defaultMod participantLimiter: sharedParticipantRequestLimiter, metrics: NewDevshardMetrics(), capacity: NewCapacityState(), - chatCache: newChatResponseCache(chatResponseCacheTTL), + chatCache: newChatResponseCache(chatResponseCacheTTL, readInt64Env("DEVSHARD_CHAT_CACHE_MAX_BYTES", defaultChatCacheMaxBytes)), suspiciousHosts: make(map[string]struct{}), settings: GatewaySettings{ DefaultModel: defaultModel, @@ -536,6 +686,10 @@ func NewManagedGateway(runtimes []*devshardRuntime, limiter *GatewayLimiter, set g.attachEscrowChecker(rt) } g.startEscrowRotatorIfEnabled() + // Settle escrows left pending by a pre-restart drain. Runs synchronously + // so the store read completes before the gateway serves traffic; the + // settlements it schedules run in their own goroutines. + g.reconcilePendingSettlements() go g.balanceCheckLoop() return g } @@ -1091,10 +1245,48 @@ func (g *Gateway) Handler() http.Handler { mux.HandleFunc("/v1/debug/signatures", g.handleSingleOnly) mux.HandleFunc("/v1/debug/signatures/collect", g.handleSingleOnly) mux.HandleFunc("/v1/debug/sync-hosts", g.handleSingleOnly) + mux.HandleFunc("/v1/debug/memstats", g.handleDebugMemStats) + // Runtime profiling, admin-gated (see isAdminPath). Mounted at the + // canonical /debug/pprof/ path so pprof.Index's sub-profile links resolve. + mux.HandleFunc("/debug/pprof/", pprof.Index) + mux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline) + mux.HandleFunc("/debug/pprof/profile", pprof.Profile) + mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol) + mux.HandleFunc("/debug/pprof/trace", pprof.Trace) mux.HandleFunc("/devshard/", g.handleDevshard) return mux } +// handleDebugMemStats reports Go runtime memory statistics so operators can +// distinguish live heap (HeapInuse) from memory the runtime has reclaimed but +// not yet returned to the OS (HeapReleased), which explains RSS that stays high +// after garbage collection. Admin-gated via the /v1/debug/ prefix. +func (g *Gateway) handleDebugMemStats(w http.ResponseWriter, r *http.Request) { + if !allowGetOrHead(w, r) { + return + } + var m runtime.MemStats + runtime.ReadMemStats(&m) + g.mu.Lock() + loadedRuntimes := len(g.runtimeOrder) + g.mu.Unlock() + writeJSON(w, map[string]any{ + "loaded_runtimes": loadedRuntimes, + "num_goroutine": runtime.NumGoroutine(), + "heap_inuse": m.HeapInuse, + "heap_alloc": m.HeapAlloc, + "heap_sys": m.HeapSys, + "heap_idle": m.HeapIdle, + "heap_released": m.HeapReleased, + "heap_objects": m.HeapObjects, + "stack_inuse": m.StackInuse, + "sys": m.Sys, + "next_gc": m.NextGC, + "num_gc": m.NumGC, + "gc_cpu_fraction": m.GCCPUFraction, + }) +} + func (g *Gateway) handlePooledModels(w http.ResponseWriter, r *http.Request) { if !allowGetOrHead(w, r) { return @@ -1262,6 +1454,28 @@ func (g *Gateway) handleDevshard(w http.ResponseWriter, r *http.Request) { rt, ok := g.runtimes[devshardID] g.mu.Unlock() if !ok { + // Non-resident devshard (inactive/settled). Read-only debug and state + // endpoints can be served from a transient runtime rehydrated from + // local storage (no chain, no host clients), then released + // immediately. Inference and other mutating paths are never + // lazy-loaded. + // + // Rehydration replays diffs and loads a snapshot into a fresh runtime. + // That is cheap per call, but an unauthenticated caller could hammer + // inactive escrow IDs to inflate gateway memory/CPU. So the hydrating + // read paths are admin-only for non-resident devshards; non-admin + // callers only get cheap registry metadata (model/active flag) that + // needs neither snapshot nor state, and everything else looks unknown. + if isReadOnlyDevshardPath(r.Method, innerPath) { + if requestHasAdminAuth(r) { + g.serveReadOnlyDevshard(w, r, devshardID, innerPath) + return + } + if g.serveInactiveDevshardMetadata(w, r, devshardID, innerPath) { + return + } + logRequestStage(ctx, "gateway_devshard_readonly_requires_admin", "escrow", devshardID, "path", innerPath) + } logRequestStage(ctx, "gateway_devshard_not_found", "escrow", devshardID) http.Error(w, fmt.Sprintf(`{"error":{"message":"unknown devshard %s"}}`, devshardID), http.StatusNotFound) return @@ -1383,6 +1597,91 @@ func (w *gatewayStatusResponseWriter) statusCode() int { return w.status } +// serveReadOnlyDevshard rehydrates a transient read-only runtime for a +// non-resident devshard, serves a single read request against it, then closes +// it. Each call loads and releases its own runtime (no caching, no +// refcounting), so concurrent reads simply build independent transient +// runtimes over the same on-disk storage. +func (g *Gateway) serveReadOnlyDevshard(w http.ResponseWriter, r *http.Request, devshardID, innerPath string) { + ctx := r.Context() + rt, known, err := g.hydrateReadOnlyRuntime(devshardID) + if err != nil { + logRequestStage(ctx, "gateway_devshard_readonly_hydrate_failed", "escrow", devshardID, "error", err) + http.Error(w, fmt.Sprintf(`{"error":{"message":"devshard %s could not be loaded: %s"}}`, devshardID, err.Error()), http.StatusBadGateway) + return + } + if !known { + logRequestStage(ctx, "gateway_devshard_not_found", "escrow", devshardID) + http.Error(w, fmt.Sprintf(`{"error":{"message":"unknown devshard %s"}}`, devshardID), http.StatusNotFound) + return + } + defer func() { + if closeErr := rt.close(); closeErr != nil { + log.Printf("gateway_devshard_readonly_close_error escrow=%s error=%v", devshardID, closeErr) + } + }() + logRequestStage(ctx, "gateway_devshard_readonly_served", "escrow", devshardID, "path", innerPath) + req := cloneRequestWithBody(r, nil) + req.URL.Path = innerPath + req.URL.RawPath = innerPath + req.RequestURI = innerPath + w.Header().Set("X-Devshard-ID", devshardID) + w.Header().Set("X-Devshard-Readonly", "1") + rt.handler.ServeHTTP(w, req) +} + +// serveInactiveDevshardMetadata answers the read-only devshard endpoints that +// can be satisfied purely from cheap registry metadata -- no snapshot or state +// load -- so an unauthenticated caller cannot use them to inflate gateway +// memory. It returns true when it handled the request; false leaves the caller +// to refuse (the devshard looks unknown to non-admins). +// +// Only /v1/models and a deliberately state-free /v1/status are +// metadata-serviceable. Every other read-only path (e.g. /v1/requests/*, and +// the admin-gated /v1/state and /v1/debug/*) needs a hydrated runtime and is +// therefore admin-only for a non-resident devshard. +func (g *Gateway) serveInactiveDevshardMetadata(w http.ResponseWriter, r *http.Request, devshardID, innerPath string) bool { + switch innerPath { + case "/v1/models", "/v1/status": + default: + return false + } + if g.store == nil { + return false + } + record, ok, err := g.store.GetDevshard(devshardID) + if err != nil || !ok { + return false + } + g.mu.Lock() + defaultModel := g.settings.DefaultModel + g.mu.Unlock() + model := firstNonEmpty(record.Model, defaultModel) + + w.Header().Set("X-Devshard-ID", devshardID) + w.Header().Set("X-Devshard-Readonly", "1") + w.Header().Set("X-Devshard-Metadata-Only", "1") + switch innerPath { + case "/v1/models": + writeModelList(w, []string{model}, RequestMaxTokensCap) + case "/v1/status": + // State-free subset only: nonce/balance/phase would require loading + // the snapshot + replaying diffs, which is exactly what we are + // refusing to do for unauthenticated callers. + writeJSON(w, map[string]any{ + "id": devshardID, + "model": model, + "active": record.Active, + "resident": false, + "settlement_pending": record.SettlementPending, + "rotation_role": record.RotationRole, + "rotation_epoch": record.RotationEpoch, + "metadata_only": true, + }) + } + return true +} + func (g *Gateway) markDevshardInactiveAfterFinalize(id string, rt *devshardRuntime) { rt.active.Store(false) if g.store == nil { @@ -1601,8 +1900,25 @@ func (g *Gateway) reserveRuntimeLocked(rt *devshardRuntime, inputTokens int64) { } func (g *Gateway) releaseRuntime(rt *devshardRuntime, inputTokens int64) { - rt.activeRequests.Add(-1) + remaining := rt.activeRequests.Add(-1) rt.reservedTokens.Add(-inputTokens) + // active=false (set during enqueue) blocks new reservations, so the count + // only drains downward — remaining==0 is the exact "last request finished" + // edge. scheduleAutoSettlement dedups, so a double-fire is harmless. + if remaining != 0 { + return + } + + if rt.settlementPending.Load() { + log.Printf("settlement_drain_complete escrow=%s reason=%s", rt.id, rt.settlementReason) + g.scheduleAutoSettlement(rt.id, rt.settlementReason) + return + } + + if rt.retirePending.Load() { + log.Printf("runtime_retire_drain_complete escrow=%s reason=%s", rt.id, rt.retireReason) + g.retireRuntime(rt.id, rt.retireReason) + } } func (rt *devshardRuntime) validateRequestedModel(requestModel string) error { @@ -2809,6 +3125,10 @@ func (g *Gateway) handleAdminDevshardAction(w http.ResponseWriter, r *http.Reque g.handleAdminDeactivateDevshard(w, r, id) return } + if len(parts) == 2 && parts[1] == "activate" && r.Method == http.MethodPost { + g.handleAdminActivateDevshard(w, r, id) + return + } if len(parts) == 2 && parts[1] == "settle" && r.Method == http.MethodPost { g.handleAdminSettleDevshard(w, r, id) return @@ -3210,24 +3530,90 @@ func (g *Gateway) handleAdminDeactivateDevshard(w http.ResponseWriter, r *http.R return } g.mu.Lock() - defer g.mu.Unlock() - rt, ok := g.runtimes[id] if !ok { + g.mu.Unlock() http.Error(w, fmt.Sprintf(`{"error":{"message":"devshard %s is not active"}}`, id), http.StatusNotFound) return } if err := g.store.SetDevshardActive(id, false); err != nil { + g.mu.Unlock() http.Error(w, fmt.Sprintf(`{"error":{"message":%q}}`, err.Error()), http.StatusInternalServerError) return } rt.active.Store(false) + // Retire from memory so the deactivated runtime stops consuming RAM. + // retireRuntimeLocked is drain-safe: if requests are still in flight it + // only marks the runtime retire-pending and returns nil, and the drain + // hook in releaseRuntime completes the retirement once the last request + // finishes. When idle it returns the runtime for us to close outside the + // lock. We hold g.mu here, so we must use the *Locked variant (the plain + // retireRuntime re-acquires g.mu and would deadlock). + retired := g.retireRuntimeLocked(id, "deactivated") + g.mu.Unlock() + if retired != nil { + if err := retired.retireClose("deactivated"); err != nil { + log.Printf("deactivate_retire_close_error escrow=%s error=%v", id, err) + } + } writeJSON(w, map[string]any{ "id": id, "active": false, }) } +// handleAdminActivateDevshard brings a non-resident (inactive) devshard back +// into the memory-resident routing pool. It builds a full runtime with chain +// access and registers it. If the devshard is already resident it simply flips +// the active flag. This is the reverse of deactivate and the recovery path for +// devshards demoted at boot or after finalize. +func (g *Gateway) handleAdminActivateDevshard(w http.ResponseWriter, r *http.Request, id string) { + if g.store == nil { + http.Error(w, `{"error":{"message":"gateway state store unavailable"}}`, http.StatusServiceUnavailable) + return + } + + g.mu.Lock() + if rt, ok := g.runtimes[id]; ok { + if err := g.store.SetDevshardActive(id, true); err != nil { + g.mu.Unlock() + http.Error(w, fmt.Sprintf(`{"error":{"message":%q}}`, err.Error()), http.StatusInternalServerError) + return + } + rt.active.Store(true) + g.mu.Unlock() + writeJSON(w, map[string]any{ + "id": id, + "active": true, + "already_resident": true, + }) + return + } + g.mu.Unlock() + + record, ok, err := g.store.GetDevshard(id) + if err != nil { + http.Error(w, fmt.Sprintf(`{"error":{"message":%q}}`, err.Error()), http.StatusInternalServerError) + return + } + if !ok { + http.Error(w, fmt.Sprintf(`{"error":{"message":"devshard %s not found"}}`, id), http.StatusNotFound) + return + } + record.Active = true + record, err = g.addCreatedEscrowRuntime(record) + if err != nil { + http.Error(w, fmt.Sprintf(`{"error":{"message":%q}}`, err.Error()), http.StatusBadGateway) + return + } + log.Printf("devshard_activated escrow=%s model=%s storage=%s", record.ID, record.Model, record.StoragePath) + writeJSON(w, map[string]any{ + "id": record.ID, + "model": record.Model, + "active": true, + }) +} + func (g *Gateway) handleAdminCleanDevshard(w http.ResponseWriter, r *http.Request, id string) { if g.store == nil { http.Error(w, `{"error":{"message":"gateway state store unavailable"}}`, http.StatusServiceUnavailable) @@ -3328,6 +3714,49 @@ func removeRuntime(runtimes []*devshardRuntime, id string) []*devshardRuntime { return out } +// retireRuntime drops a runtime from the in-memory registry and closes it, +// releasing its user session and the per-runtime SQLite handles that session owns. +func (g *Gateway) retireRuntime(id, reason string) bool { + g.mu.Lock() + rt := g.retireRuntimeLocked(id, reason) + g.mu.Unlock() + if rt == nil { + return false + } + // Close the per-runtime SQLite store outside g.mu: it is disk I/O and must + // not block other gateway operations that contend for the lock. retireClose + // also flushes a final snapshot so the frozen escrow rebuilds replay-free. + if err := rt.retireClose(reason); err != nil { + log.Printf("runtime_retire_close_error escrow=%s reason=%q error=%v", id, reason, err) + } + log.Printf("runtime_retired escrow=%s reason=%q", id, reason) + return true +} + +// retireRuntimeLocked removes the runtime from the registry and returns it so +// the caller can close it outside the lock. It returns nil when nothing was +// retired: the runtime is unknown, or its retirement was deferred because +// requests are still in flight. Callers must hold g.mu. +func (g *Gateway) retireRuntimeLocked(id, reason string) *devshardRuntime { + rt, ok := g.runtimes[id] + if !ok { + log.Printf("runtime_retire_skipped escrow=%s reason=%q cause=not_registered", id, reason) + return nil + } + if inFlight := rt.activeRequests.Load(); inFlight > 0 { + rt.retireReason = reason + rt.retirePending.Store(true) + log.Printf("runtime_retire_deferred escrow=%s reason=%q active_requests=%d", id, reason, inFlight) + return nil + } + delete(g.runtimes, id) + g.runtimeOrder = removeRuntime(g.runtimeOrder, id) + if g.capacity != nil { + g.capacity.RemoveEscrow(id) + } + return rt +} + func (g *Gateway) sortRuntimeOrderLocked() { slices.SortFunc(g.runtimeOrder, func(a, b *devshardRuntime) int { return strings.Compare(a.id, b.id) @@ -3352,12 +3781,15 @@ func (g *Gateway) attachEscrowChecker(rt *devshardRuntime) { rt.proxy.redundancy.onEscrowMissing = func() { go g.escrowChecker.TriggerCheck(escrowID, func() { g.deactivateDevshardByID(escrowID) + // Escrow no longer exists on chain -- nothing to settle. + g.retireRuntime(escrowID, "escrow confirmed missing on chain") }) } } rt.proxy.redundancy.onBalanceExhausted = func() { if !g.escrowRotationEnabled() { g.deactivateDevshardByIDWithReason(escrowID, "escrow balance exhausted") + g.retireRuntime(escrowID, "escrow balance exhausted") return } log.Printf("gateway_replacing_exhausted_escrow escrow=%s", escrowID) @@ -3398,24 +3830,116 @@ func (g *Gateway) deactivateDevshardByIDWithReason(id, reason string) bool { return true } +// deactivateAndSettleDevshardByID stops new traffic to an escrow and settles +// it. If requests are still in flight it marks the escrow settlement-pending +// and returns; the drain hook in releaseRuntime settles once the last request +// finishes. Otherwise it settles immediately. func (g *Gateway) deactivateAndSettleDevshardByID(id, reason string) { if !g.deactivateDevshardByIDWithReason(id, reason) { return } + g.markSettlementPending(id, reason) + + g.mu.Lock() + rt, ok := g.runtimes[id] + g.mu.Unlock() + if ok && rt.activeRequests.Load() > 0 { + log.Printf("settlement_queued_waiting_for_drain escrow=%s reason=%s active_requests=%d", + id, reason, rt.activeRequests.Load()) + return + } g.scheduleAutoSettlement(id, reason) } +// markSettlementPending records that an escrow must be settled once its +// in-flight requests drain. The reason is stored before the flag so the +// lock-free drain hook in releaseRuntime reads a consistent value. +func (g *Gateway) markSettlementPending(id, reason string) { + g.mu.Lock() + rt, ok := g.runtimes[id] + if ok { + rt.settlementReason = reason + rt.settlementPending.Store(true) + } + g.mu.Unlock() + if g.store != nil { + if err := g.store.SetDevshardSettlementPending(id, true); err != nil { + log.Printf("settlement_pending_persist_failed escrow=%s error=%v", id, err) + } + } +} + +// reconcilePendingSettlements settles escrows that were marked pending before +// a restart. After a restart no requests are in flight, so each such escrow +// can settle immediately. Hydrates the in-memory marker too. +func (g *Gateway) reconcilePendingSettlements() { + if g.store == nil { + return + } + state, ok, err := g.store.LoadState() + if err != nil || !ok { + if err != nil { + log.Printf("settlement_reconcile_load_failed error=%v", err) + } + return + } + // Honor the operator's config: when settlement is disabled, never settle on + // startup. Leave the marker intact so a later re-enable still settles it. + if !state.Settings.EscrowRotation.SettlementEnabled { + for _, devshard := range state.Devshards { + if !devshard.Active && devshard.SettlementPending { + log.Printf("settlement_reconcile_skipped escrow=%s reason=settlement_disabled", devshard.ID) + } + } + return + } + for _, devshard := range state.Devshards { + if devshard.Active || !devshard.SettlementPending { + continue + } + g.mu.Lock() + if rt, exists := g.runtimes[devshard.ID]; exists { + rt.settlementReason = "startup_reconcile" + rt.settlementPending.Store(true) + } + g.mu.Unlock() + // Non-resident pending devshards are still settled: scheduleAutoSettlement + // drives settleDevshardOnChain, which rehydrates a transient full runtime + // from local storage when the devshard is not resident in memory. + log.Printf("settlement_reconcile_queued escrow=%s", devshard.ID) + g.scheduleAutoSettlement(devshard.ID, "startup_reconcile") + } +} + +// clearSettlementPending is called after a successful settlement so a +// restart-time reconcile does not re-settle the escrow. +func (g *Gateway) clearSettlementPending(id string) { + g.mu.Lock() + rt, ok := g.runtimes[id] + if ok { + rt.settlementPending.Store(false) + } + g.mu.Unlock() + if g.store != nil { + if err := g.store.SetDevshardSettlementPending(id, false); err != nil { + log.Printf("settlement_pending_clear_failed escrow=%s error=%v", id, err) + } + } +} + func (g *Gateway) retireRotatedDevshard(ctx context.Context, id, reason string, settings GatewaySettings) (bool, error) { if !settings.EscrowRotation.SettlementEnabled { if g.deactivateDevshardByIDWithReason(id, reason) { log.Printf("escrow_rotation_deactivated_without_settlement escrow=%s reason=%q", id, reason) } + g.retireRuntime(id, reason) return false, nil } log.Printf("escrow_rotation_settling escrow=%s reason=%q", id, reason) if _, err := gatewaySettleDevshardOnChain(g, ctx, id, adminSettleEscrowRequest{}); err != nil { return false, err } + g.retireRuntime(id, reason) return true, nil } @@ -3480,6 +4004,7 @@ func (g *Gateway) replaceDepletedEscrow(ctx context.Context, id, modelID, reason id, result.EscrowID, model.ModelID, reason, result.TxHash) if !settings.EscrowRotation.SettlementEnabled { g.deactivateDevshardByIDWithReason(id, reason) + g.retireRuntime(id, reason) } else { g.deactivateAndSettleDevshardByID(id, reason) } @@ -3530,11 +4055,16 @@ func (g *Gateway) scheduleAutoSettlement(id, reason string) { if err == nil { log.Printf("auto_settle_submitted escrow=%s reason=%s tx_hash=%s settler=%s", id, reason, result.TxHash, result.Settler) + g.retireRuntime(id, reason) return } log.Printf("auto_settle_failed escrow=%s reason=%s attempt=%d/%d error=%v", id, reason, attempt, autoSettlementMaxAttempts, err) if attempt == autoSettlementMaxAttempts { + // Settlement exhausted its retries; free the in-memory runtime + // anyway so a permanently-unsettleable escrow cannot leak its + // SQLite store. On-disk state is preserved for manual recovery. + g.retireRuntime(id, reason) return } time.Sleep(autoSettlementRetryInterval) diff --git a/devshard/cmd/devshardctl/gateway_inactive_access_test.go b/devshard/cmd/devshardctl/gateway_inactive_access_test.go new file mode 100644 index 0000000000..19f46873ed --- /dev/null +++ b/devshard/cmd/devshardctl/gateway_inactive_access_test.go @@ -0,0 +1,124 @@ +package main + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "path/filepath" + "testing" + + "github.com/stretchr/testify/require" +) + +// newInactiveDevshardGateway builds a store-backed gateway that knows about a +// single non-resident (not in memory), inactive devshard "77". Because no +// runtime is registered, any /devshard/77/... read takes the non-resident +// branch of handleDevshard. +func newInactiveDevshardGateway(t *testing.T) *Gateway { + t.Helper() + store, err := NewGatewayStore(filepath.Join(t.TempDir(), "gateway.db")) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, store.Close()) }) + + settings := GatewaySettings{DefaultModel: "m"} + devshards := []GatewayDevshardState{{ + RuntimeConfig: RuntimeConfig{ + ID: "77", + PrivateKeyHex: "secret", + Model: "m", + StoragePath: filepath.Join(t.TempDir(), "escrow-77"), + }, + Active: false, + }} + require.NoError(t, store.Initialize(settings, devshards)) + + return NewManagedGateway(nil, NewGatewayLimiter(0, 0), settings, t.TempDir(), store) +} + +// A non-admin caller may read a non-resident devshard's /v1/status, but only +// the cheap, state-free metadata subset -- no snapshot/state is loaded. +func TestInactiveDevshardPublicStatusIsMetadataOnly(t *testing.T) { + g := newInactiveDevshardGateway(t) + + req := httptest.NewRequest(http.MethodGet, "/devshard/77/v1/status", nil) + rec := httptest.NewRecorder() + g.handleDevshard(rec, req) + + require.Equal(t, http.StatusOK, rec.Code) + require.Equal(t, "1", rec.Header().Get("X-Devshard-Metadata-Only")) + + var body map[string]any + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &body)) + require.Equal(t, "77", body["id"]) + require.Equal(t, "m", body["model"]) + require.Equal(t, false, body["active"]) + require.Equal(t, false, body["resident"]) + require.Equal(t, true, body["metadata_only"]) + + // Fields that would require replaying diffs / loading a snapshot must be + // absent: the whole point is that no state was hydrated. + require.NotContains(t, body, "nonce") + require.NotContains(t, body, "balance") + require.NotContains(t, body, "phase") +} + +// /v1/models is derivable from cheap registry config, so it is public for a +// non-resident devshard. +func TestInactiveDevshardPublicModelsIsMetadataOnly(t *testing.T) { + g := newInactiveDevshardGateway(t) + + req := httptest.NewRequest(http.MethodGet, "/devshard/77/v1/models", nil) + rec := httptest.NewRecorder() + g.handleDevshard(rec, req) + + require.Equal(t, http.StatusOK, rec.Code) + require.Equal(t, "1", rec.Header().Get("X-Devshard-Metadata-Only")) + require.Contains(t, rec.Body.String(), `"m"`) +} + +// A read that needs hydrated state (here /v1/requests/*) is refused for a +// non-admin caller on a non-resident devshard: it must look unknown and must +// not hydrate (no metadata-only response either). +func TestInactiveDevshardPublicStatefulReadRefused(t *testing.T) { + g := newInactiveDevshardGateway(t) + + req := httptest.NewRequest(http.MethodGet, "/devshard/77/v1/requests/abc", nil) + rec := httptest.NewRecorder() + g.handleDevshard(rec, req) + + require.Equal(t, http.StatusNotFound, rec.Code) + require.Empty(t, rec.Header().Get("X-Devshard-Metadata-Only")) + require.Contains(t, rec.Body.String(), "unknown devshard") +} + +// With admin auth present, the non-resident read path hydrates a transient +// runtime instead of returning the metadata-only response. We only assert it +// did NOT take the metadata-only branch (hydration may succeed or fail +// depending on on-disk storage, but either way it is not metadata-only). +func TestInactiveDevshardAdminReadHydrates(t *testing.T) { + g := newInactiveDevshardGateway(t) + + req := httptest.NewRequest(http.MethodGet, "/devshard/77/v1/status", nil) + req = req.WithContext(context.WithValue(req.Context(), adminAuthContextKey{}, true)) + rec := httptest.NewRecorder() + g.handleDevshard(rec, req) + + require.Empty(t, rec.Header().Get("X-Devshard-Metadata-Only"), + "admin read must hydrate, not serve the metadata-only response") +} + +// An unknown (not-in-store) devshard must not reveal anything to a non-admin, +// even on the otherwise-public metadata paths. +func TestUnknownDevshardPublicReadIsNotFound(t *testing.T) { + g := newInactiveDevshardGateway(t) + + for _, path := range []string{"/devshard/does-not-exist/v1/status", "/devshard/does-not-exist/v1/models"} { + req := httptest.NewRequest(http.MethodGet, path, nil) + rec := httptest.NewRecorder() + g.handleDevshard(rec, req) + + require.Equal(t, http.StatusNotFound, rec.Code, "path=%s", path) + require.Empty(t, rec.Header().Get("X-Devshard-Metadata-Only"), "path=%s", path) + } +} diff --git a/devshard/cmd/devshardctl/gateway_runtime_retire_test.go b/devshard/cmd/devshardctl/gateway_runtime_retire_test.go new file mode 100644 index 0000000000..680706b11c --- /dev/null +++ b/devshard/cmd/devshardctl/gateway_runtime_retire_test.go @@ -0,0 +1,149 @@ +package main + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// newRetireTestGateway builds a minimal Gateway holding a single active runtime +// registered in both the lookup map and the ordered slice, mirroring how the +// real registry is populated. +func newRetireTestGateway(id string) (*Gateway, *devshardRuntime) { + rt := &devshardRuntime{id: id} + rt.active.Store(true) + g := &Gateway{ + runtimes: map[string]*devshardRuntime{id: rt}, + runtimeOrder: []*devshardRuntime{rt}, + rotationFailures: make(map[string]struct{}), + } + return g, rt +} + +// TestRetireRuntimeRemovesRuntimeFromRegistry pins the core leak fix: retiring a +// runtime must drop it from both g.runtimes and g.runtimeOrder so its +// user.Session (and the per-runtime SQLite handles it owns) can be released. +func TestRetireRuntimeRemovesRuntimeFromRegistry(t *testing.T) { + g, _ := newRetireTestGateway("12") + + require.True(t, g.retireRuntime("12", "test")) + + _, stillRegistered := g.runtimes["12"] + require.False(t, stillRegistered, "runtime must be removed from g.runtimes") + require.Empty(t, g.runtimeOrder, "runtime must be removed from g.runtimeOrder") + + // Idempotent: retiring an already-gone runtime is a no-op, not a panic. + require.False(t, g.retireRuntime("12", "test")) +} + +// TestRetireRuntimeDefersWhileRequestsInFlight guards against closing a SQLite +// store out from under an in-flight request: retirement must defer (and leave +// the runtime registered) until the request count drains to zero. +func TestRetireRuntimeDefersWhileRequestsInFlight(t *testing.T) { + g, rt := newRetireTestGateway("12") + rt.activeRequests.Store(1) + + require.False(t, g.retireRuntime("12", "busy")) + _, stillRegistered := g.runtimes["12"] + require.True(t, stillRegistered, "busy runtime must stay registered") + require.True(t, rt.retirePending.Load(), "deferred retire must record its intent") + require.Equal(t, "busy", rt.retireReason) + + rt.activeRequests.Store(0) + require.True(t, g.retireRuntime("12", "drained")) + _, stillRegistered = g.runtimes["12"] + require.False(t, stillRegistered) +} + +// TestReleaseRuntimeRetiresAfterDrain: a retire deferred while busy fires once +// the last request drains through releaseRuntime. +func TestReleaseRuntimeRetiresAfterDrain(t *testing.T) { + g, rt := newRetireTestGateway("12") + rt.activeRequests.Store(1) + + require.False(t, g.retireRuntime("12", "balance exhausted")) + _, stillRegistered := g.runtimes["12"] + require.True(t, stillRegistered, "busy runtime must stay registered") + + g.releaseRuntime(rt, 0) + + _, stillRegistered = g.runtimes["12"] + require.False(t, stillRegistered, "drained runtime must be retired by releaseRuntime") + require.Empty(t, g.runtimeOrder) +} + +// TestReleaseRuntimeRetiresWithOnlyRetirePending exercises the retire branch in +// isolation: only retirePending set (no settlement), drain → retire. +func TestReleaseRuntimeRetiresWithOnlyRetirePending(t *testing.T) { + g, rt := newRetireTestGateway("12") + rt.activeRequests.Store(1) + rt.retireReason = "balance exhausted" + rt.retirePending.Store(true) + + g.releaseRuntime(rt, 0) + + _, stillRegistered := g.runtimes["12"] + require.False(t, stillRegistered, "retire branch must fire on drain") + require.Empty(t, g.runtimeOrder) +} + +// TestReleaseRuntimeDefersWhileRequestsRemain: while remaining != 0 nothing +// fires; settled stays 0 until the last request drains. Uses settlementPending +// because scheduleAutoSettlement fires regardless of the live count, so a +// broken guard leaks as settled>0 (retire would self-defer and hide it). +func TestReleaseRuntimeDefersWhileRequestsRemain(t *testing.T) { + rt := gatewayTestRuntimeForLimits(t, "12", balanceMinimumThreshold-1, nonceDeactivationLimit-1) + g, _, settled := gatewayTestDepletionGateway(t, rt) + + g.reserveRuntime(rt, 1) + g.reserveRuntime(rt, 1) + rt.settlementReason = "low_balance" + rt.settlementPending.Store(true) + + g.releaseRuntime(rt, 1) // remaining == 1 → quiet + require.Never(t, func() bool { return settled.Load() > 0 }, 200*time.Millisecond, 20*time.Millisecond) + + g.releaseRuntime(rt, 1) // remaining == 0 → settles once + require.Eventually(t, func() bool { return settled.Load() == 1 }, time.Second, 10*time.Millisecond) +} + +// TestRetireRotatedDevshardRetiresWithoutSettlement covers the no-settle +// terminal path: when settlement is disabled, the rotated-out runtime is +// deactivated AND retired in the same step. +func TestRetireRotatedDevshardRetiresWithoutSettlement(t *testing.T) { + g, _ := newRetireTestGateway("12") + settings := GatewaySettings{EscrowRotation: EscrowRotationSettings{SettlementEnabled: false}} + + settled, err := g.retireRotatedDevshard(context.Background(), "12", "rotated", settings) + require.NoError(t, err) + require.False(t, settled) + + _, stillRegistered := g.runtimes["12"] + require.False(t, stillRegistered, "no-settle rotation must retire the runtime") +} + +// TestRetireRotatedDevshardRetiresAfterSettlement covers the settle terminal +// path: the runtime stays alive through settlement (which reads its session) +// and is retired only once settlement succeeds. +func TestRetireRotatedDevshardRetiresAfterSettlement(t *testing.T) { + g, _ := newRetireTestGateway("12") + settings := GatewaySettings{EscrowRotation: EscrowRotationSettings{SettlementEnabled: true}} + + oldSettle := gatewaySettleDevshardOnChain + gatewaySettleDevshardOnChain = func(g *Gateway, _ context.Context, id string, _ adminSettleEscrowRequest) (*SettleDevshardEscrowResult, error) { + // The session must still be reachable at settlement time. + _, ok := g.runtimes[id] + require.True(t, ok, "runtime must still be registered during settlement") + return &SettleDevshardEscrowResult{TxHash: "OK"}, nil + } + t.Cleanup(func() { gatewaySettleDevshardOnChain = oldSettle }) + + settled, err := g.retireRotatedDevshard(context.Background(), "12", "rotated", settings) + require.NoError(t, err) + require.True(t, settled) + + _, stillRegistered := g.runtimes["12"] + require.False(t, stillRegistered, "settled rotation must retire the runtime") +} diff --git a/devshard/cmd/devshardctl/gateway_store.go b/devshard/cmd/devshardctl/gateway_store.go index f8fb2e05b3..906d903c7c 100644 --- a/devshard/cmd/devshardctl/gateway_store.go +++ b/devshard/cmd/devshardctl/gateway_store.go @@ -285,11 +285,12 @@ func normalizeGatewayModelAccess(access []GatewayModelAccessSettings) []GatewayM type GatewayDevshardState struct { RuntimeConfig - Active bool `json:"active"` - RotationRole string `json:"rotation_role,omitempty"` - RotationEpoch uint64 `json:"rotation_epoch,omitempty"` - CreatedAt string `json:"created_at,omitempty"` - UpdatedAt string `json:"updated_at,omitempty"` + Active bool `json:"active"` + SettlementPending bool `json:"settlement_pending,omitempty"` + RotationRole string `json:"rotation_role,omitempty"` + RotationEpoch uint64 `json:"rotation_epoch,omitempty"` + CreatedAt string `json:"created_at,omitempty"` + UpdatedAt string `json:"updated_at,omitempty"` } type GatewaySuspiciousHost struct { @@ -375,6 +376,7 @@ func NewGatewayStore(path string) (*GatewayStore, error) { active INTEGER NOT NULL DEFAULT 1, rotation_role TEXT NOT NULL DEFAULT '', rotation_epoch INTEGER NOT NULL DEFAULT 0, + settlement_pending INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL, updated_at TEXT NOT NULL )`, @@ -437,6 +439,10 @@ func NewGatewayStore(path string) (*GatewayStore, error) { db.Close() return nil, fmt.Errorf("migrate gateway devshard epoch: %w", err) } + if err := ensureGatewayDevshardsColumn(db, "settlement_pending", "INTEGER NOT NULL DEFAULT 0"); err != nil { + db.Close() + return nil, fmt.Errorf("migrate gateway devshard settlement_pending: %w", err) + } if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS participant_throttle_state ( participant_key TEXT PRIMARY KEY, tokens REAL NOT NULL DEFAULT 0, @@ -626,7 +632,7 @@ func (s *GatewayStore) LoadState() (GatewayState, bool, error) { rows, err := s.db.Query(` SELECT id, private_key_hex, private_key_env, model, storage_path, active, created_at, updated_at, protocol_version, - rotation_role, rotation_epoch + rotation_role, rotation_epoch, settlement_pending FROM gateway_devshards ORDER BY id`) if err != nil { @@ -636,6 +642,7 @@ func (s *GatewayStore) LoadState() (GatewayState, bool, error) { for rows.Next() { var devshard GatewayDevshardState var active int + var settlementPending int if err := rows.Scan( &devshard.ID, &devshard.PrivateKeyHex, @@ -648,10 +655,12 @@ func (s *GatewayStore) LoadState() (GatewayState, bool, error) { &devshard.ProtocolVersion, &devshard.RotationRole, &devshard.RotationEpoch, + &settlementPending, ); err != nil { return GatewayState{}, false, fmt.Errorf("scan gateway devshard: %w", err) } devshard.Active = active != 0 + devshard.SettlementPending = settlementPending != 0 state.Devshards = append(state.Devshards, devshard) } if err := rows.Err(); err != nil { @@ -1080,12 +1089,21 @@ func (s *GatewayStore) UpsertDevshard(devshard GatewayDevshardState) error { func (s *GatewayStore) upsertDevshardTx(tx *sql.Tx, devshard GatewayDevshardState, now string) error { createdAt := now - _ = tx.QueryRow(`SELECT created_at FROM gateway_devshards WHERE id = ?`, devshard.ID).Scan(&createdAt) + if err := tx.QueryRow(`SELECT created_at FROM gateway_devshards WHERE id = ?`, devshard.ID).Scan(&createdAt); err != nil && err != sql.ErrNoRows { + return fmt.Errorf("lookup created_at for devshard %s: %w", devshard.ID, err) + } + // Preserve the existing settlement_pending marker so an unrelated upsert + // never silently clears a queued settlement; a brand-new row falls back + // to the value carried on devshard. + settlementPending := gatewayBoolToInt(devshard.SettlementPending) + if err := tx.QueryRow(`SELECT settlement_pending FROM gateway_devshards WHERE id = ?`, devshard.ID).Scan(&settlementPending); err != nil && err != sql.ErrNoRows { + return fmt.Errorf("lookup settlement_pending for devshard %s: %w", devshard.ID, err) + } if _, err := tx.Exec(` INSERT OR REPLACE INTO gateway_devshards ( id, private_key_hex, private_key_env, model, storage_path, active, created_at, updated_at, protocol_version, - rotation_role, rotation_epoch - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + rotation_role, rotation_epoch, settlement_pending + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, strings.TrimSpace(devshard.ID), strings.TrimSpace(devshard.PrivateKeyHex), strings.TrimSpace(devshard.PrivateKeyEnv), @@ -1097,12 +1115,51 @@ func (s *GatewayStore) upsertDevshardTx(tx *sql.Tx, devshard GatewayDevshardStat strings.TrimSpace(devshard.ProtocolVersion), strings.TrimSpace(devshard.RotationRole), devshard.RotationEpoch, + settlementPending, ); err != nil { return fmt.Errorf("upsert gateway devshard %s: %w", devshard.ID, err) } return nil } +// GetDevshard returns the registry record for a single devshard. The second +// return value is false when no row exists for the id. It is used by lazy +// hydration to look up the config of a non-resident devshard without loading +// the entire registry. +func (s *GatewayStore) GetDevshard(id string) (GatewayDevshardState, bool, error) { + id = strings.TrimSpace(id) + var devshard GatewayDevshardState + var active int + var settlementPending int + err := s.db.QueryRow(` + SELECT id, private_key_hex, private_key_env, model, storage_path, active, created_at, updated_at, protocol_version, + rotation_role, rotation_epoch, settlement_pending + FROM gateway_devshards + WHERE id = ?`, id).Scan( + &devshard.ID, + &devshard.PrivateKeyHex, + &devshard.PrivateKeyEnv, + &devshard.Model, + &devshard.StoragePath, + &active, + &devshard.CreatedAt, + &devshard.UpdatedAt, + &devshard.ProtocolVersion, + &devshard.RotationRole, + &devshard.RotationEpoch, + &settlementPending, + ) + if err == sql.ErrNoRows { + return GatewayDevshardState{}, false, nil + } + if err != nil { + return GatewayDevshardState{}, false, fmt.Errorf("get devshard %s: %w", id, err) + } + devshard.Active = active != 0 + devshard.SettlementPending = settlementPending != 0 + return devshard, true, nil +} + func (s *GatewayStore) SetDevshardActive(id string, active bool) error { res, err := s.db.Exec(` UPDATE gateway_devshards @@ -1125,6 +1182,28 @@ func (s *GatewayStore) SetDevshardActive(id string, active bool) error { return nil } +func (s *GatewayStore) SetDevshardSettlementPending(id string, pending bool) error { + res, err := s.db.Exec(` + UPDATE gateway_devshards + SET settlement_pending = ?, updated_at = ? + WHERE id = ?`, + gatewayBoolToInt(pending), + time.Now().UTC().Format(time.RFC3339Nano), + strings.TrimSpace(id), + ) + if err != nil { + return fmt.Errorf("update devshard %s settlement_pending=%t: %w", id, pending, err) + } + n, err := res.RowsAffected() + if err != nil { + return fmt.Errorf("rows affected for devshard %s: %w", id, err) + } + if n == 0 { + return fmt.Errorf("devshard %s not found", id) + } + return nil +} + func (s *GatewayStore) DeleteDevshard(id string) error { res, err := s.db.Exec(`DELETE FROM gateway_devshards WHERE id = ?`, strings.TrimSpace(id)) if err != nil { diff --git a/devshard/cmd/devshardctl/gateway_store_test.go b/devshard/cmd/devshardctl/gateway_store_test.go index c61371e1e1..9b150ba8a3 100644 --- a/devshard/cmd/devshardctl/gateway_store_test.go +++ b/devshard/cmd/devshardctl/gateway_store_test.go @@ -789,6 +789,49 @@ func TestEscrowRotationUsesEpochSwitchHeightDuringPoC(t *testing.T) { require.Equal(t, 1, settleAttempts) } +func TestGatewayStoreSetDevshardSettlementPending(t *testing.T) { + store, err := NewGatewayStore(filepath.Join(t.TempDir(), "gateway.db")) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, store.Close()) }) + + require.NoError(t, store.Initialize(GatewaySettings{ + ChainREST: "http://node:1317", DefaultModel: "m", DefaultRequestMaxTokens: 1000, + }.WithTuningDefaults(), []GatewayDevshardState{{ + RuntimeConfig: RuntimeConfig{ID: "12", PrivateKeyHex: "secret", Model: "m"}, + Active: true, + }})) + + // Default is not pending. + state, ok, err := store.LoadState() + require.NoError(t, err) + require.True(t, ok) + require.False(t, gatewayDevshardsByID(state.Devshards)["12"].SettlementPending) + + // Set pending → persisted and survives reload. + require.NoError(t, store.SetDevshardSettlementPending("12", true)) + state, _, err = store.LoadState() + require.NoError(t, err) + require.True(t, gatewayDevshardsByID(state.Devshards)["12"].SettlementPending) + + // An unrelated upsert must NOT wipe the pending marker. + require.NoError(t, store.UpsertDevshard(GatewayDevshardState{ + RuntimeConfig: RuntimeConfig{ID: "12", PrivateKeyHex: "secret", Model: "m"}, + Active: false, + })) + state, _, err = store.LoadState() + require.NoError(t, err) + require.True(t, gatewayDevshardsByID(state.Devshards)["12"].SettlementPending) + + // Clear pending. + require.NoError(t, store.SetDevshardSettlementPending("12", false)) + state, _, err = store.LoadState() + require.NoError(t, err) + require.False(t, gatewayDevshardsByID(state.Devshards)["12"].SettlementPending) + + // Unknown id errors. + require.Error(t, store.SetDevshardSettlementPending("nope", true)) +} + func gatewayDevshardsByID(devshards []GatewayDevshardState) map[string]GatewayDevshardState { byID := make(map[string]GatewayDevshardState, len(devshards)) for _, devshard := range devshards { diff --git a/devshard/cmd/devshardctl/gateway_test.go b/devshard/cmd/devshardctl/gateway_test.go index d488acd1cd..9c9d707917 100644 --- a/devshard/cmd/devshardctl/gateway_test.go +++ b/devshard/cmd/devshardctl/gateway_test.go @@ -93,6 +93,21 @@ func seedGatewayTestCapacity(g *Gateway, weights map[string]float64) { g.capacity.SetHostWeightViews(seeded, seeded, nil, nil) } +// withSettleDevshardOnChainSideEffects wraps a settlement stub so successful +// calls apply the same post-broadcast bookkeeping as settleDevshardOnChain. +func withSettleDevshardOnChainSideEffects( + stub func(g *Gateway, ctx context.Context, id string, req adminSettleEscrowRequest) (*SettleDevshardEscrowResult, error), +) func(*Gateway, context.Context, string, adminSettleEscrowRequest) (*SettleDevshardEscrowResult, error) { + return func(g *Gateway, ctx context.Context, id string, req adminSettleEscrowRequest) (*SettleDevshardEscrowResult, error) { + result, err := stub(g, ctx, id, req) + if err != nil { + return nil, err + } + g.clearSettlementPending(id) + return result, nil + } +} + func gatewayTestDepletionGateway(t *testing.T, rt *devshardRuntime, modifySettings ...func(*GatewaySettings)) (*Gateway, *atomic.Int32, *atomic.Int32) { t.Helper() @@ -138,11 +153,11 @@ func gatewayTestDepletionGateway(t *testing.T, rt *devshardRuntime, modifySettin created.Add(1) return &CreateDevshardEscrowResult{EscrowID: 99, TxHash: "OK"}, nil } - gatewaySettleDevshardOnChain = func(_ *Gateway, _ context.Context, id string, _ adminSettleEscrowRequest) (*SettleDevshardEscrowResult, error) { + gatewaySettleDevshardOnChain = withSettleDevshardOnChainSideEffects(func(_ *Gateway, _ context.Context, id string, _ adminSettleEscrowRequest) (*SettleDevshardEscrowResult, error) { require.Equal(t, rt.id, id) settled.Add(1) return &SettleDevshardEscrowResult{EscrowID: mustParseUintForTest(t, id), TxHash: "SETTLED", Settler: "settler"}, nil - } + }) t.Cleanup(func() { gatewayCreateDepletionEscrow = oldCreate gatewaySettleDevshardOnChain = oldSettle @@ -238,6 +253,93 @@ func TestGatewayCheckBalancesKeepsRuntimeBelowLimits(t *testing.T) { require.True(t, rt.active.Load()) } +func TestEnqueueSettlementWaitsForActiveRequests(t *testing.T) { + rt := gatewayTestRuntimeForLimits(t, "12", balanceMinimumThreshold-1, nonceDeactivationLimit-1) + g, _, settled := gatewayTestDepletionGateway(t, rt) + + // One request in flight → settlement must NOT fire yet, but escrow is + // deactivated and marked pending (in-memory + persisted). + g.reserveRuntime(rt, 1) + g.deactivateAndSettleDevshardByID("12", "low_balance") + + require.False(t, rt.active.Load()) + require.True(t, rt.settlementPending.Load()) + state, ok, err := g.store.LoadState() + require.NoError(t, err) + require.True(t, ok) + require.True(t, gatewayDevshardsByID(state.Devshards)["12"].SettlementPending) + require.EqualValues(t, 0, settled.Load()) + + // Draining the last request triggers exactly one settlement, which + // clears the marker. + g.releaseRuntime(rt, 1) + require.Eventually(t, func() bool { + return settled.Load() == 1 && !rt.settlementPending.Load() + }, time.Second, 10*time.Millisecond) + + state, _, err = g.store.LoadState() + require.NoError(t, err) + require.False(t, gatewayDevshardsByID(state.Devshards)["12"].SettlementPending) +} + +func TestEnqueueSettlementSettlesImmediatelyWhenDrained(t *testing.T) { + rt := gatewayTestRuntimeForLimits(t, "12", balanceMinimumThreshold-1, nonceDeactivationLimit-1) + g, _, settled := gatewayTestDepletionGateway(t, rt) + + // No active requests → settle right away. + g.deactivateAndSettleDevshardByID("12", "low_balance") + + require.Eventually(t, func() bool { + return settled.Load() == 1 && !rt.active.Load() + }, time.Second, 10*time.Millisecond) +} + +func TestReconcilePendingSettlementsSettlesDrainedEscrow(t *testing.T) { + rt := gatewayTestRuntimeForLimits(t, "12", balanceMinimumThreshold, nonceDeactivationLimit-1) + g, _, settled := gatewayTestDepletionGateway(t, rt) + + // Simulate a marker left behind by a pre-restart drain. + rt.active.Store(false) + require.NoError(t, g.store.SetDevshardActive("12", false)) + require.NoError(t, g.store.SetDevshardSettlementPending("12", true)) + + g.reconcilePendingSettlements() + + require.Eventually(t, func() bool { + return settled.Load() == 1 && !rt.settlementPending.Load() + }, time.Second, 10*time.Millisecond) +} + +func TestReconcilePendingSettlementsSkipsActiveOrUnflagged(t *testing.T) { + rt := gatewayTestRuntimeForLimits(t, "12", balanceMinimumThreshold, nonceDeactivationLimit-1) + g, _, settled := gatewayTestDepletionGateway(t, rt) + + // Active escrow, no pending marker → nothing to do. + g.reconcilePendingSettlements() + require.Never(t, func() bool { return settled.Load() > 0 }, 200*time.Millisecond, 20*time.Millisecond) +} + +func TestReconcilePendingSettlementsSkipsWhenSettlementDisabled(t *testing.T) { + rt := gatewayTestRuntimeForLimits(t, "12", balanceMinimumThreshold, nonceDeactivationLimit-1) + g, _, settled := gatewayTestDepletionGateway(t, rt, func(settings *GatewaySettings) { + settings.EscrowRotation.SettlementEnabled = false + }) + + // Inactive escrow flagged pending, but settlement is disabled → reconcile + // must not settle, and the marker is preserved for a later re-enable. + rt.active.Store(false) + require.NoError(t, g.store.SetDevshardActive("12", false)) + require.NoError(t, g.store.SetDevshardSettlementPending("12", true)) + + g.reconcilePendingSettlements() + + require.Never(t, func() bool { return settled.Load() > 0 }, 200*time.Millisecond, 20*time.Millisecond) + state, ok, err := g.store.LoadState() + require.NoError(t, err) + require.True(t, ok) + require.True(t, gatewayDevshardsByID(state.Devshards)["12"].SettlementPending) +} + func TestParseDevshardPath(t *testing.T) { id, inner, ok := parseDevshardPath("/devshard/12/v1/debug/perf") require.True(t, ok) diff --git a/devshard/cmd/devshardctl/main.go b/devshard/cmd/devshardctl/main.go index 003672eabb..b71ca84243 100644 --- a/devshard/cmd/devshardctl/main.go +++ b/devshard/cmd/devshardctl/main.go @@ -363,16 +363,25 @@ func mustBuildGateway(gatewayStore *GatewayStore, gatewayState GatewayState, bas } func buildGatewayRuntimes(gatewayStore *GatewayStore, gatewayState *GatewayState, baseStorageDir string, perf *PerfTracker) ([]*devshardRuntime, error) { - // Load ALL devshards (active and inactive) so that inactive ones - // remain accessible for finalization, debug, and settlement retrieval. - // Inactive runtimes are loaded with active=false and excluded from - // the inference routing pool. + // Load only ACTIVE devshards at boot. Inactive devshards (deactivated, + // finalized, or settled) stay in the registry but are not built into + // memory-resident runtimes: keeping hundreds of dormant escrows resident + // wastes RAM, and probing each one against the chain at startup causes a + // boot-time request storm. Inactive devshards are rehydrated on demand: + // read-only from local storage for debug/state endpoints, or fully (with + // chain access) for manual settlement. See hydrateReadOnlyRuntime and the + // lazy settle path. type cfgEntry struct { cfg RuntimeConfig active bool } allEntries := make([]cfgEntry, 0, len(gatewayState.Devshards)) + skippedInactive := 0 for _, devshard := range gatewayState.Devshards { + if !devshard.Active { + skippedInactive++ + continue + } allEntries = append(allEntries, cfgEntry{cfg: devshard.RuntimeConfig, active: devshard.Active}) } allCfgs := make([]RuntimeConfig, len(allEntries)) @@ -470,8 +479,8 @@ func buildGatewayRuntimes(gatewayStore *GatewayStore, gatewayState *GatewayState out = append(out, rt) } } - log.Printf("build_runtimes_parallel count=%d active=%d inactive=%d skipped=%d total_elapsed_ms=%d", - len(out), activeCount, inactiveCount, len(skipped), time.Since(t0).Milliseconds()) + log.Printf("build_runtimes_parallel count=%d active=%d inactive=%d skipped=%d skipped_inactive=%d total_elapsed_ms=%d", + len(out), activeCount, inactiveCount, len(skipped), skippedInactive, time.Since(t0).Milliseconds()) return out, nil } @@ -562,6 +571,7 @@ func isAuthExemptPath(path string) bool { func isAdminPath(path string) bool { if strings.HasPrefix(path, "/v1/admin/") || strings.HasPrefix(path, "/v1/debug/") || + strings.HasPrefix(path, "/debug/pprof/") || path == "/v1/finalize" || path == "/v1/state" { return true diff --git a/devshard/cmd/devshardctl/proxy.go b/devshard/cmd/devshardctl/proxy.go index d050acd715..23c5be6e3a 100644 --- a/devshard/cmd/devshardctl/proxy.go +++ b/devshard/cmd/devshardctl/proxy.go @@ -758,20 +758,23 @@ func (p *Proxy) handleDebugPairwise(w http.ResponseWriter, r *http.Request) { } func (p *Proxy) handleDebugState(w http.ResponseWriter, r *http.Request) { - st := p.sm.SnapshotState() + // Live counts come from InferenceStatusCounts (computed under the read + // lock without deep-copying records); sealed records are evicted from the + // live map, so they are reported separately from the seal-nonce index. + liveTotal, statusCounts := p.sm.InferenceStatusCounts() sealed := p.sm.ExportSealedNonces() - liveStatusCounts := make(map[string]int) - for _, rec := range st.Inferences { - name := inferenceStatusName[rec.Status] + liveStatusCounts := make(map[string]int, len(statusCounts)) + for status, n := range statusCounts { + name := inferenceStatusName[status] if name == "" { - name = fmt.Sprintf("unknown(%d)", rec.Status) + name = fmt.Sprintf("unknown(%d)", status) } - liveStatusCounts[name]++ + liveStatusCounts[name] = n } phaseStr := "active" - switch st.Phase { + switch p.sm.Phase() { case types.PhaseFinalizing: phaseStr = "finalizing" case types.PhaseSettlement: @@ -779,14 +782,14 @@ func (p *Proxy) handleDebugState(w http.ResponseWriter, r *http.Request) { } writeJSON(w, map[string]any{ - "nonce": st.LatestNonce, + "nonce": p.sm.LatestNonce(), "phase": phaseStr, - "balance": st.Balance, - "live_inferences": len(st.Inferences), + "balance": p.sm.Balance(), + "live_inferences": liveTotal, "sealed_inferences": len(sealed), "live_status_counts": liveStatusCounts, // Deprecated: same as live_inferences; kept for older scripts. - "total_inferences": len(st.Inferences), + "total_inferences": liveTotal, "status_counts": liveStatusCounts, }) } @@ -810,13 +813,12 @@ func (p *Proxy) handleStatus(w http.ResponseWriter, r *http.Request) { phaseStr = fmt.Sprintf("unknown(%d)", phase) } - st := p.sm.SnapshotState() - cfg := st.Config + cfg := p.sm.Config() status := statusResponse{ EscrowID: p.escrowID, Nonce: p.session.Nonce(), Phase: phaseStr, - Balance: st.Balance, + Balance: p.sm.Balance(), Config: statusSessionConfig{ RefusalTimeout: cfg.RefusalTimeout, ExecutionTimeout: cfg.ExecutionTimeout, @@ -1054,8 +1056,12 @@ func (p *Proxy) handleSyncHosts(w http.ResponseWriter, r *http.Request) { }) } +// handleState returns the escrow summary: session scalars, config, group, and +// the small per-slot maps. It deliberately omits the (potentially large) +// inference map -- that lives at /v1/debug/inferences -- so this endpoint stays +// bounded regardless of how many inferences an escrow has accumulated. func (p *Proxy) handleState(w http.ResponseWriter, r *http.Request) { - st := p.sm.SnapshotState() + st := p.sm.SnapshotStateNoInferences() var phaseStr string switch st.Phase { @@ -1092,33 +1098,6 @@ func (p *Proxy) handleState(w http.ResponseWriter, r *http.Request) { } } - allInferences := p.sm.ExportAllInferenceRecords() - inferences := make(map[string]any, len(allInferences)) - for id, rec := range allInferences { - name := inferenceStatusName[rec.Status] - if name == "" { - name = fmt.Sprintf("unknown(%d)", rec.Status) - } - inferences[fmt.Sprintf("%d", id)] = map[string]any{ - "status": name, - "executor_slot": rec.ExecutorSlot, - "model": rec.Model, - "prompt_hash": hex.EncodeToString(rec.PromptHash), - "response_hash": hex.EncodeToString(rec.ResponseHash), - "input_length": rec.InputLength, - "max_tokens": rec.MaxTokens, - "input_tokens": rec.InputTokens, - "output_tokens": rec.OutputTokens, - "reserved_cost": rec.ReservedCost, - "actual_cost": rec.ActualCost, - "started_at": rec.StartedAt, - "confirmed_at": rec.ConfirmedAt, - "votes_valid": rec.VotesValid, - "votes_invalid": rec.VotesInvalid, - "validated_by": rec.ValidatedBy.SetBits(), - } - } - hostStats := make(map[string]any, len(st.HostStats)) for slot, hs := range st.HostStats { hostStats[fmt.Sprintf("%d", slot)] = map[string]any{ @@ -1140,7 +1119,6 @@ func (p *Proxy) handleState(w http.ResponseWriter, r *http.Request) { resp := map[string]any{ "session": session, "group": group, - "inferences": inferences, "host_stats": hostStats, "revealed_seeds": revealedSeeds, "warm_keys": warmKeys, @@ -1148,3 +1126,42 @@ func (p *Proxy) handleState(w http.ResponseWriter, r *http.Request) { writeJSON(w, resp) } + +// handleDebugInferences returns the full inference map for the escrow. It is a +// debug-only endpoint (potentially large: up to the per-escrow inference cap) +// split out of /v1/state so that summary reads stay cheap. It uses +// ExportAllInferenceRecords rather than the live map so records already +// sealed to storage still appear in the dump. +func (p *Proxy) handleDebugInferences(w http.ResponseWriter, r *http.Request) { + inferenceMap := p.sm.ExportAllInferenceRecords() + inferences := make(map[string]any, len(inferenceMap)) + for id, rec := range inferenceMap { + name := inferenceStatusName[rec.Status] + if name == "" { + name = fmt.Sprintf("unknown(%d)", rec.Status) + } + inferences[fmt.Sprintf("%d", id)] = map[string]any{ + "status": name, + "executor_slot": rec.ExecutorSlot, + "model": rec.Model, + "prompt_hash": hex.EncodeToString(rec.PromptHash), + "response_hash": hex.EncodeToString(rec.ResponseHash), + "input_length": rec.InputLength, + "max_tokens": rec.MaxTokens, + "input_tokens": rec.InputTokens, + "output_tokens": rec.OutputTokens, + "reserved_cost": rec.ReservedCost, + "actual_cost": rec.ActualCost, + "started_at": rec.StartedAt, + "confirmed_at": rec.ConfirmedAt, + "votes_valid": rec.VotesValid, + "votes_invalid": rec.VotesInvalid, + "validated_by": rec.ValidatedBy.SetBits(), + } + } + + writeJSON(w, map[string]any{ + "total_inferences": len(inferences), + "inferences": inferences, + }) +} diff --git a/devshard/cmd/devshardctl/proxy_test.go b/devshard/cmd/devshardctl/proxy_test.go index cf1ad8c99e..b00a96059d 100644 --- a/devshard/cmd/devshardctl/proxy_test.go +++ b/devshard/cmd/devshardctl/proxy_test.go @@ -1711,19 +1711,30 @@ func TestHandleState_IncludesSealedInferences(t *testing.T) { proxy := &Proxy{sm: sm, escrowID: escrowID} - req := httptest.NewRequest(http.MethodGet, "/v1/state", nil) + // The full dump lives at /v1/debug/inferences (moved out of /v1/state so + // summary reads stay bounded); sealed records must still appear there. + req := httptest.NewRequest(http.MethodGet, "/v1/debug/inferences", nil) rec := httptest.NewRecorder() - proxy.handleState(rec, req) + proxy.handleDebugInferences(rec, req) require.Equal(t, http.StatusOK, rec.Code) - var stateResp map[string]any - require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &stateResp)) - inferences, ok := stateResp["inferences"].(map[string]any) - require.True(t, ok, "/v1/state must expose inferences map") + var dumpResp map[string]any + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &dumpResp)) + inferences, ok := dumpResp["inferences"].(map[string]any) + require.True(t, ok, "/v1/debug/inferences must expose inferences map") inf, ok := inferences["1"].(map[string]any) - require.True(t, ok, "sealed inference 1 must appear in /v1/state") + require.True(t, ok, "sealed inference 1 must appear in /v1/debug/inferences") require.Equal(t, "finished", inf["status"]) require.Equal(t, "llama", inf["model"]) + + // /v1/state stays bounded: no inference map in the summary. + stateReq := httptest.NewRequest(http.MethodGet, "/v1/state", nil) + stateRec := httptest.NewRecorder() + proxy.handleState(stateRec, stateReq) + require.Equal(t, http.StatusOK, stateRec.Code) + var stateResp map[string]any + require.NoError(t, json.Unmarshal(stateRec.Body.Bytes(), &stateResp)) + require.NotContains(t, stateResp, "inferences", "/v1/state must not carry the inference dump") } func TestProxyStatusIncludesChainPhaseSnapshot(t *testing.T) { diff --git a/devshard/cmd/devshardctl/response_cache.go b/devshard/cmd/devshardctl/response_cache.go index 33efd26aa0..ecdb6058c9 100644 --- a/devshard/cmd/devshardctl/response_cache.go +++ b/devshard/cmd/devshardctl/response_cache.go @@ -13,10 +13,29 @@ import ( const chatResponseCacheTTL = time.Hour +// chatCacheSweepInterval bounds how often Set pays for a full expiry sweep. +// Expiry used to be enforced only lazily inside Get for the exact key being +// looked up; since keys are hashes of full request bodies, unique requests +// were never looked up again and their entries lived until process restart. +const chatCacheSweepInterval = time.Minute + +// defaultChatCacheMaxBytes caps the total body bytes held by the cache +// (overridable via DEVSHARD_CHAT_CACHE_MAX_BYTES). The cap is a safety net +// against traffic bursts within the TTL window; the sweep handles steady +// state. +const defaultChatCacheMaxBytes = int64(256 << 20) + +// chatCacheEntryOverhead approximates the per-entry cost beyond the body: +// map bucket, key string (64-hex sha256), and struct fields. +const chatCacheEntryOverhead = 256 + type chatResponseCache struct { - mu sync.Mutex - ttl time.Duration - entries map[string]cachedChatResponse + mu sync.Mutex + ttl time.Duration + maxBytes int64 + entries map[string]cachedChatResponse + totalBytes int64 + lastSweep time.Time } type cachedChatResponse struct { @@ -29,13 +48,50 @@ type cachedChatResponse struct { ExpiresAt time.Time } -func newChatResponseCache(ttl time.Duration) *chatResponseCache { +func newChatResponseCache(ttl time.Duration, maxBytes int64) *chatResponseCache { if ttl <= 0 { ttl = chatResponseCacheTTL } + if maxBytes <= 0 { + maxBytes = defaultChatCacheMaxBytes + } return &chatResponseCache{ - ttl: ttl, - entries: make(map[string]cachedChatResponse), + ttl: ttl, + maxBytes: maxBytes, + entries: make(map[string]cachedChatResponse), + } +} + +func chatCacheEntrySize(entry cachedChatResponse) int64 { + return int64(len(entry.Body)+len(entry.ContentType)+len(entry.EscrowID)+len(entry.SourceRequestID)) + chatCacheEntryOverhead +} + +// deleteLocked removes key from the map and adjusts the byte total. +// Caller must hold c.mu. +func (c *chatResponseCache) deleteLocked(key string) { + entry, ok := c.entries[key] + if !ok { + return + } + delete(c.entries, key) + c.totalBytes -= chatCacheEntrySize(entry) + if c.totalBytes < 0 { + // Entries written directly in tests bypass accounting. + c.totalBytes = 0 + } +} + +// sweepExpiredLocked scans the whole map and drops entries whose TTL has +// passed, at most once per chatCacheSweepInterval. Caller must hold c.mu. +func (c *chatResponseCache) sweepExpiredLocked(now time.Time) { + if now.Sub(c.lastSweep) < chatCacheSweepInterval { + return + } + c.lastSweep = now + for key, entry := range c.entries { + if !entry.ExpiresAt.After(now) { + c.deleteLocked(key) + } } } @@ -58,11 +114,11 @@ func (c *chatResponseCache) Get(key string, now time.Time) (cachedChatResponse, return cachedChatResponse{}, false } if !entry.ExpiresAt.After(now) { - delete(c.entries, key) + c.deleteLocked(key) return cachedChatResponse{}, false } if responseBodyHasNonCacheableError(entry.Body) { - delete(c.entries, key) + c.deleteLocked(key) return cachedChatResponse{}, false } entry.Body = append([]byte(nil), entry.Body...) @@ -83,7 +139,33 @@ func (c *chatResponseCache) Set(key string, entry cachedChatResponse, now time.T c.mu.Lock() defer c.mu.Unlock() + c.sweepExpiredLocked(now) + + c.deleteLocked(key) // drop any previous version's byte count c.entries[key] = entry + c.totalBytes += chatCacheEntrySize(entry) + + // Size cap: evict arbitrary entries (map iteration order) until under + // the limit. This is a dedup cache -- evicting a "wrong" entry only + // costs one cache miss, so eviction order isn't worth tracking. + for other := range c.entries { + if c.totalBytes <= c.maxBytes { + break + } + if other != key { + c.deleteLocked(other) + } + } +} + +// Stats reports the current entry count and approximate retained bytes. +func (c *chatResponseCache) Stats() (entryCount int, totalBytes int64) { + if c == nil { + return 0, 0 + } + c.mu.Lock() + defer c.mu.Unlock() + return len(c.entries), c.totalBytes } func serveCachedChatResponse(w http.ResponseWriter, r *http.Request, entry cachedChatResponse) { diff --git a/devshard/cmd/devshardctl/response_cache_test.go b/devshard/cmd/devshardctl/response_cache_test.go index 8912bdec87..3f80366818 100644 --- a/devshard/cmd/devshardctl/response_cache_test.go +++ b/devshard/cmd/devshardctl/response_cache_test.go @@ -92,8 +92,98 @@ func TestGatewayChatCacheCaptureRejectsRuntimeAndCapabilityErrors(t *testing.T) } } +func okBody(marker byte, size int) []byte { + filler := make([]byte, size) + for i := range filler { + filler[i] = 'a' + (marker+byte(i))%26 + } + return []byte(`{"choices":[{"message":{"content":"` + string(filler) + `"}}]}`) +} + +func cacheEntryForTest(marker byte, bodySize int) cachedChatResponse { + return cachedChatResponse{ + EscrowID: "escrow-1", + StatusCode: http.StatusOK, + Body: okBody(marker, bodySize), + } +} + +func TestChatResponseCacheSweepsExpiredEntriesOnSet(t *testing.T) { + cache := newChatResponseCache(time.Minute, 0) + start := time.Now() + + cache.Set("old-1", cacheEntryForTest(1, 10), start) + cache.Set("old-2", cacheEntryForTest(2, 10), start) + + count, _ := cache.Stats() + require.Equal(t, 2, count) + + // A Set past both the TTL and the sweep interval must remove the + // expired entries even though their keys are never looked up again. + cache.Set("new", cacheEntryForTest(3, 10), start.Add(2*time.Minute)) + + count, _ = cache.Stats() + require.Equal(t, 1, count) + _, ok := cache.entries["old-1"] + require.False(t, ok) + _, ok = cache.entries["old-2"] + require.False(t, ok) + _, ok = cache.entries["new"] + require.True(t, ok) +} + +func TestChatResponseCacheEvictsWhenOverByteCap(t *testing.T) { + // Cap fits roughly two 4KB entries plus overhead, not three. + cache := newChatResponseCache(time.Minute, 10_000) + now := time.Now() + + cache.Set("a", cacheEntryForTest(1, 4096), now) + cache.Set("b", cacheEntryForTest(2, 4096), now) + cache.Set("c", cacheEntryForTest(3, 4096), now) + + _, totalBytes := cache.Stats() + require.LessOrEqual(t, totalBytes, int64(10_000)) + + // The just-inserted entry must survive eviction. + _, ok := cache.entries["c"] + require.True(t, ok) +} + +func TestChatResponseCacheOverwriteDoesNotLeakBytes(t *testing.T) { + cache := newChatResponseCache(time.Minute, 1<<20) + now := time.Now() + + for i := 0; i < 100; i++ { + cache.Set("same-key", cacheEntryForTest(byte(i), 4096), now) + } + + count, totalBytes := cache.Stats() + require.Equal(t, 1, count) + require.Less(t, totalBytes, int64(2*4096+2*chatCacheEntryOverhead)) + + entry, ok := cache.Get("same-key", now) + require.True(t, ok) + require.Equal(t, string(okBody(99, 4096)), string(entry.Body)) +} + +func TestChatResponseCacheGetDeletesExpiredAndAdjustsBytes(t *testing.T) { + cache := newChatResponseCache(time.Minute, 0) + now := time.Now() + + cache.Set("k", cacheEntryForTest(1, 128), now) + _, totalBytes := cache.Stats() + require.Greater(t, totalBytes, int64(0)) + + _, ok := cache.Get("k", now.Add(2*time.Minute)) + require.False(t, ok) + + count, totalBytes := cache.Stats() + require.Equal(t, 0, count) + require.Equal(t, int64(0), totalBytes) +} + func TestChatResponseCacheDropsPreviouslyCachedNonCacheableErrors(t *testing.T) { - cache := newChatResponseCache(time.Minute) + cache := newChatResponseCache(time.Minute, 0) cache.entries["bad"] = cachedChatResponse{ EscrowID: "escrow-1", StatusCode: http.StatusBadGateway, diff --git a/devshard/state/machine.go b/devshard/state/machine.go index 76bc434bb2..506fb1b0e8 100644 --- a/devshard/state/machine.go +++ b/devshard/state/machine.go @@ -494,6 +494,14 @@ func (sm *StateMachine) Balance() uint64 { return sm.state.Balance } +// Config returns a copy of the session config (a small value type). Use this +// instead of SnapshotState().Config to avoid deep-copying the inference map. +func (sm *StateMachine) Config() types.SessionConfig { + sm.mu.RLock() + defer sm.mu.RUnlock() + return sm.state.Config +} + // SnapshotState returns a deep copy of the current escrow state. func (sm *StateMachine) SnapshotState() types.EscrowState { sm.mu.RLock() @@ -501,6 +509,48 @@ func (sm *StateMachine) SnapshotState() types.EscrowState { return *cloneEscrowState(sm.state) } +// SnapshotStateNoInferences returns a deep copy of the escrow state with the +// (potentially large) inference map omitted. All other fields, including the +// small per-slot maps, are copied. Use it for summary/state endpoints that do +// not render individual inference records, avoiding the cost of copying up to +// tens of thousands of them. +func (sm *StateMachine) SnapshotStateNoInferences() types.EscrowState { + sm.mu.RLock() + defer sm.mu.RUnlock() + src := sm.state + // Shallow struct copy; SealedAcc ([]byte) is shared deliberately: it is + // only ever replaced wholesale (append to a nil slice), never mutated in + // place, so readers of the snapshot see a stable value. + s := *src + s.Inferences = nil + + s.Group = make([]types.SlotAssignment, len(src.Group)) + copy(s.Group, src.Group) + + s.HostStats = make(map[uint32]*types.HostStats, len(src.HostStats)) + for k, v := range src.HostStats { + cp := *v + s.HostStats[k] = &cp + } + + s.WarmKeys = make(map[uint32]string, len(src.WarmKeys)) + maps.Copy(s.WarmKeys, src.WarmKeys) + + return s +} + +// InferenceStatusCounts returns the total number of inferences and a per-status +// breakdown, computed under the read lock without deep-copying any records. +func (sm *StateMachine) InferenceStatusCounts() (int, map[types.InferenceStatus]int) { + sm.mu.RLock() + defer sm.mu.RUnlock() + counts := make(map[types.InferenceStatus]int) + for _, rec := range sm.state.Inferences { + counts[rec.Status]++ + } + return len(sm.state.Inferences), counts +} + // ExportState returns a deep-copied pointer form used by recovery snapshots. func (sm *StateMachine) ExportState() *types.EscrowState { sm.mu.RLock() diff --git a/devshard/user/flush_snapshot_test.go b/devshard/user/flush_snapshot_test.go new file mode 100644 index 0000000000..2191598b9a --- /dev/null +++ b/devshard/user/flush_snapshot_test.go @@ -0,0 +1,173 @@ +package user + +import ( + "context" + "sync" + "testing" + + "github.com/stretchr/testify/require" + + "devshard/host" + "devshard/internal/testutil" + "devshard/signing" + "devshard/state" + "devshard/storage" + "devshard/stub" + "devshard/types" +) + +// replaySpyStore wraps a storage.Storage and records every GetDiffs range so a +// test can distinguish diff replay from catch-up backfill. During recovery, +// replayed diffs (fed through ApplyLocal to rebuild SM state) come from a +// GetDiffs call whose range starts after the snapshot nonce; backfill calls +// end at the snapshot nonce. +type replaySpyStore struct { + storage.Storage + mu sync.Mutex + calls []spyGetDiffs +} + +type spyGetDiffs struct { + from, to uint64 + count int +} + +func (s *replaySpyStore) GetDiffs(escrowID string, from, to uint64) ([]types.DiffRecord, error) { + recs, err := s.Storage.GetDiffs(escrowID, from, to) + s.mu.Lock() + s.calls = append(s.calls, spyGetDiffs{from: from, to: to, count: len(recs)}) + s.mu.Unlock() + return recs, err +} + +// replayedRecords sums records returned for calls whose range starts strictly +// after snapNonce -- exactly the post-snapshot diffs RecoverSession replays. +func (s *replaySpyStore) replayedRecords(snapNonce uint64) int { + s.mu.Lock() + defer s.mu.Unlock() + total := 0 + for _, c := range s.calls { + if c.from > snapNonce { + total += c.count + } + } + return total +} + +// buildLiveSession builds a storage-backed session and its state machine, +// mirroring setupRecoverableSession but returning the live session so a test +// can drive it and then flush a snapshot. +func buildLiveSession( + t *testing.T, numHosts int, store storage.Storage, +) (*Session, *state.StateMachine, []types.SlotAssignment, []*signing.Secp256k1Signer, *signing.Secp256k1Signer) { + t.Helper() + hosts := make([]*signing.Secp256k1Signer, numHosts) + for i := range hosts { + hosts[i] = testutil.MustGenerateKey(t) + } + user := testutil.MustGenerateKey(t) + group := testutil.MakeGroup(hosts) + config := testutil.DefaultConfig(numHosts) + verifier := signing.NewSecp256k1Verifier() + + require.NoError(t, store.CreateSession(storage.CreateSessionParams{ + EscrowID: "escrow-1", + Version: testutil.RuntimeTestVersion, + CreatorAddr: user.Address(), + Config: config, + Group: group, + InitialBalance: 100000, + })) + + clients := make([]HostClient, numHosts) + for i := range hosts { + sm := newTestStateMachine(t, "escrow-1", config, group, 100000, user.Address(), verifier) + h, err := host.NewHost(sm, hosts[i], stub.NewInferenceEngine(), "escrow-1", group, nil, host.WithGrace(10)) + require.NoError(t, err) + clients[i] = &InProcessClient{Host: h} + } + + userSM := newTestStateMachine(t, "escrow-1", config, group, 100000, user.Address(), verifier) + session, err := NewSession(userSM, user, "escrow-1", group, clients, verifier, WithStorage(store)) + require.NoError(t, err) + + return session, userSM, group, hosts, user +} + +// TestFlushSnapshot_RetiredEscrowRebuildsWithoutReplay is the core guarantee of +// the retire-time snapshot flush: an escrow whose nonce advanced past the last +// periodic snapshot (here: none, since numInferences < snapshotInterval) must, +// after FlushSnapshot, rebuild via RecoverSession with zero diff replay. +func TestFlushSnapshot_RetiredEscrowRebuildsWithoutReplay(t *testing.T) { + store := newTestStore(t) + numHosts := 3 + numInferences := 5 // < snapshotInterval, so no periodic snapshot is taken + + session, liveSM, group, hosts, user := buildLiveSession(t, numHosts, store) + + ctx := context.Background() + params := InferenceParams{ + Model: "llama", Prompt: testutil.TestPrompt, + InputLength: 100, MaxTokens: 50, StartedAt: 1000, + } + for i := 0; i < numInferences; i++ { + _, err := session.SendInference(ctx, params) + require.NoError(t, err) + } + require.Equal(t, uint64(numInferences), session.Nonce()) + + // No periodic snapshot exists yet: a plain rebuild here would replay all + // numInferences diffs. + _, _, err := store.LoadSnapshot("escrow-1") + require.ErrorIs(t, err, storage.ErrSnapshotNotFound) + + // Retire: flush a final snapshot at the current (frozen) nonce. + require.NoError(t, session.FlushSnapshot()) + + snapNonce, _, err := store.LoadSnapshot("escrow-1") + require.NoError(t, err) + require.Equal(t, uint64(numInferences), snapNonce, "flush must snapshot at the frozen nonce") + + // Rebuild through a spy to observe replay. With a snapshot at LatestNonce, + // RecoverSession must take the early-return path and never fetch (let alone + // replay) any post-snapshot diff. + verifier := signing.NewSecp256k1Verifier() + spy := &replaySpyStore{Storage: store} + rec, recSM, err := RecoverSession(spy, user, verifier, "escrow-1", testutil.RuntimeTestVersion, group, buildRecoveryClients(t, hosts, group, user)) + require.NoError(t, err) + require.Equal(t, uint64(numInferences), rec.Nonce()) + require.Zero(t, spy.replayedRecords(snapNonce), "retired escrow must rebuild with zero diff replay") + + // The rebuilt state must be identical to the live session's state. + recRoot, err := recSM.ComputeStateRoot() + require.NoError(t, err) + liveRoot, err := liveSM.ComputeStateRoot() + require.NoError(t, err) + require.Equal(t, liveRoot, recRoot, "flushed snapshot must reproduce the live state root") +} + +// TestFlushSnapshot_NoStoreOrEmptyIsNoop verifies FlushSnapshot is safe on +// sessions with nothing to persist: no store configured, or nonce still 0. +func TestFlushSnapshot_NoStoreOrEmptyIsNoop(t *testing.T) { + // nonce == 0 (no inferences) with a store: must not write a snapshot. + store := newTestStore(t) + session, _, _, _, _ := buildLiveSession(t, 3, store) + require.Equal(t, uint64(0), session.Nonce()) + require.NoError(t, session.FlushSnapshot()) + _, _, err := store.LoadSnapshot("escrow-1") + require.ErrorIs(t, err, storage.ErrSnapshotNotFound, "flush at nonce 0 must not write a snapshot") + + // No store configured: flush is a no-op that returns nil. + verifier := signing.NewSecp256k1Verifier() + hostSigner := testutil.MustGenerateKey(t) + user := testutil.MustGenerateKey(t) + group := testutil.MakeGroup([]*signing.Secp256k1Signer{hostSigner}) + config := testutil.DefaultConfig(1) + hostSM := newTestStateMachine(t, "escrow-1", config, group, 100000, user.Address(), verifier) + h, err := host.NewHost(hostSM, hostSigner, stub.NewInferenceEngine(), "escrow-1", group, nil, host.WithGrace(10)) + require.NoError(t, err) + sm := newTestStateMachine(t, "escrow-1", config, group, 100000, user.Address(), verifier) + noStore, err := NewSession(sm, user, "escrow-1", group, []HostClient{&InProcessClient{Host: h}}, verifier) + require.NoError(t, err) + require.NoError(t, noStore.FlushSnapshot()) +} diff --git a/devshard/user/httpsession.go b/devshard/user/httpsession.go index bbddb9edcb..b827a22f84 100644 --- a/devshard/user/httpsession.go +++ b/devshard/user/httpsession.go @@ -5,6 +5,7 @@ import ( "fmt" "os" "path/filepath" + "strings" devshardpkg "devshard" "devshard/bridge" @@ -38,6 +39,62 @@ func resolveHTTPSessionStoragePath(escrowID, configured string) string { return filepath.Join(home, ".cache", "gonka", fmt.Sprintf("devshard-%s", escrowID)) } +// LocalSessionConfig holds the parameters needed to rehydrate a user Session +// entirely from local storage, with no chain access and no host clients. +type LocalSessionConfig struct { + PrivateKeyHex string + EscrowID string + StoragePath string + ProtocolVersion types.ProtocolVersion +} + +// NewLocalSession rehydrates a Session from local SQLite storage without +// contacting the chain and without wiring any host clients. The returned +// session can answer read-only queries (state, status, debug, settlement +// build) but cannot dispatch new inferences. Callers own the returned +// Session and must Close it when done, which also closes the underlying +// storage handle. +// +// Warm-key verification is intentionally omitted (nil resolver): stored +// diffs carry their warm-key deltas, which RecoverSession injects before +// replay, so no chain-backed resolver is needed to rebuild state. +func NewLocalSession(cfg LocalSessionConfig) (*Session, *state.StateMachine, error) { + if strings.TrimSpace(cfg.StoragePath) == "" { + return nil, nil, fmt.Errorf("local session requires a storage path") + } + signer, err := signing.SignerFromHex(cfg.PrivateKeyHex) + if err != nil { + return nil, nil, fmt.Errorf("create signer: %w", err) + } + pv := cfg.ProtocolVersion + if pv == "" { + pv = types.ProtocolV1 + } + verifier := signing.NewSecp256k1Verifier() + version := devshardpkg.ProtocolSessionVersion(pv) + + store, err := storage.NewSQLite(cfg.StoragePath) + if err != nil { + return nil, nil, fmt.Errorf("open storage: %w", err) + } + meta, err := store.GetSessionMeta(cfg.EscrowID) + if err != nil { + store.Close() + return nil, nil, fmt.Errorf("get session meta: %w", err) + } + // No host clients: read-only sessions never dispatch inferences. The + // slice length must match the group so NewSession's invariant holds. + clients := make([]HostClient, len(meta.Group)) + session, sm, err := RecoverSession(store, signer, verifier, cfg.EscrowID, version, meta.Group, clients, + state.WithProtocolVersion(pv), + ) + if err != nil { + store.Close() + return nil, nil, fmt.Errorf("recover session: %w", err) + } + return session, sm, nil +} + // NewHTTPSession creates a user Session wired with HTTP clients to real dapi hosts. // It queries the bridge for escrow and group info, then creates transport clients // for each slot. diff --git a/devshard/user/recover.go b/devshard/user/recover.go index cfe5400cdf..44b9993466 100644 --- a/devshard/user/recover.go +++ b/devshard/user/recover.go @@ -292,16 +292,24 @@ func saveSnapshot(store storage.Storage, sm *state.StateMachine, escrowID string // or state-machine locks held -- this is what enables async background // snapshots from the runtime hot path). func writeSnapshot(store storage.Storage, escrowID string, nonce uint64, state *types.EscrowState, cursor map[int]uint64) { + _ = writeSnapshotErr(store, escrowID, nonce, state, cursor) +} + +// writeSnapshotErr is writeSnapshot with an error return, for synchronous +// callers (e.g. Session.FlushSnapshot on retire) that want to know whether the +// snapshot landed. It logs on failure exactly like writeSnapshot. +func writeSnapshotErr(store storage.Storage, escrowID string, nonce uint64, state *types.EscrowState, cursor map[int]uint64) error { blob := sessionSnapshot{State: state, HostSyncNonce: cursor} data, err := json.Marshal(blob) if err != nil { log.Printf("recover_session escrow=%s snapshot_marshal_failed=%v", escrowID, err) - return + return err } if err := store.SaveSnapshot(escrowID, nonce, data); err != nil { log.Printf("recover_session escrow=%s snapshot_save_failed=%v", escrowID, err) - return + return err } log.Printf("recover_session escrow=%s snapshot_saved nonce=%d size_bytes=%d host_cursors=%d", escrowID, nonce, len(data), len(cursor)) + return nil } diff --git a/devshard/user/session.go b/devshard/user/session.go index 9b1a00a69d..14a8eb0ec1 100644 --- a/devshard/user/session.go +++ b/devshard/user/session.go @@ -672,6 +672,61 @@ func (s *Session) maybeSaveSnapshotLocked() { }() } +// FlushSnapshot synchronously persists a state snapshot at the current nonce. +// It is called when a runtime is retired from memory (deactivate, settle, +// rotation) so the now-frozen escrow can later be rebuilt -- for a read-only +// debug/state request or a reactivation -- without replaying the diff tail +// accumulated since the last periodic (every snapshotInterval) snapshot. +// +// Periodic snapshots are only taken on snapshotInterval boundaries, so a +// retired escrow otherwise carries up to snapshotInterval-1 un-snapshotted +// diffs that every rebuild would replay. This flush captures the final nonce +// once, at the lifecycle transition, making all later rebuilds replay-free. +// +// It serializes against the async periodic writer via snapshotInFlight so a +// slow in-flight periodic save cannot land after (and overwrite) this final +// snapshot with a staler nonce. Safe to call once, at retire; a no-op when +// no diffs have been applied (nonce == 0) or no store is configured. +func (s *Session) FlushSnapshot() error { + if s.store == nil { + return nil + } + // Acquire the snapshot slot, waiting briefly for any in-flight async + // save to finish. Bounded so retire never blocks indefinitely; if the + // async writer is still busy after the wait we proceed anyway (SQLite + // serializes the writes, and retire runs after drain so this is rare). + acquired := false + for i := 0; i < 200; i++ { + if s.snapshotInFlight.CompareAndSwap(false, true) { + acquired = true + break + } + time.Sleep(10 * time.Millisecond) + } + if acquired { + defer s.snapshotInFlight.Store(false) + } + + s.mu.Lock() + if s.nonce == 0 { + s.mu.Unlock() + return nil + } + // Deep-copy state and cursor under the session lock, mirroring + // maybeSaveSnapshotLocked, then release before the disk write. + stateCopy := s.sm.ExportState() + cursor := make(map[int]uint64, len(s.hostSyncNonce)) + for k, v := range s.hostSyncNonce { + cursor[k] = v + } + nonce := s.nonce + store := s.store + escrowID := s.escrowID + s.mu.Unlock() + + return writeSnapshotErr(store, escrowID, nonce, stateCopy, cursor) +} + // PrepareInference composes a diff, applies it locally, advances nonce, // and returns everything needed for the HTTP send. Thread-safe. // @@ -1692,7 +1747,7 @@ func (s *Session) IsNonceFinished(nonce uint64) bool { // sendTime is when the nonce's network call started. func (s *Session) HandleTimeout(ctx context.Context, nonce uint64, sendTime time.Time, payload *host.InferencePayload) (TimeoutResult, error) { s.mu.Lock() - cfg := s.sm.SnapshotState().Config + cfg := s.sm.Config() confirmedAt := int64(0) if o, ok := s.nonceStates[nonce]; ok { confirmedAt = o.confirmedAt diff --git a/devshard/user/session_close_test.go b/devshard/user/session_close_test.go new file mode 100644 index 0000000000..ae26e3f7b4 --- /dev/null +++ b/devshard/user/session_close_test.go @@ -0,0 +1,70 @@ +package user + +import ( + "testing" + + "github.com/stretchr/testify/require" + + "devshard/storage" + "devshard/types" +) + +// closeCountingStore is a storage.Storage that records how many times Close is +// called. Every other method is an inert stub: this fake exists only to prove +// that Session.Close releases the underlying store, which is the resource the +// per-runtime memory leak was failing to free. +type closeCountingStore struct { + closeCalls int +} + +func (s *closeCountingStore) CreateSession(storage.CreateSessionParams) error { return nil } +func (s *closeCountingStore) MarkSettled(string) error { return nil } +func (s *closeCountingStore) ListActiveSessions() ([]storage.ActiveSession, error) { + return nil, nil +} +func (s *closeCountingStore) AppendDiff(string, types.DiffRecord) error { return nil } +func (s *closeCountingStore) GetDiffs(string, uint64, uint64) ([]types.DiffRecord, error) { + return nil, nil +} +func (s *closeCountingStore) AddSignature(string, uint64, uint32, []byte) error { return nil } +func (s *closeCountingStore) GetSignatures(string, uint64) (map[uint32][]byte, error) { + return nil, nil +} +func (s *closeCountingStore) GetSessionMeta(string) (*storage.SessionMeta, error) { + return nil, storage.ErrSessionNotFound +} +func (s *closeCountingStore) MarkFinalized(string, uint64) error { return nil } +func (s *closeCountingStore) LastFinalized(string) (uint64, error) { return 0, nil } +func (s *closeCountingStore) SaveSnapshot(string, uint64, []byte) error { return nil } +func (s *closeCountingStore) LoadSnapshot(string) (uint64, []byte, error) { + return 0, nil, storage.ErrSnapshotNotFound +} +func (s *closeCountingStore) PruneEpoch(uint64) error { return nil } +func (s *closeCountingStore) DeleteSealedInferences(string) error { return nil } +func (s *closeCountingStore) InsertSealedInference(string, storage.InferenceRow) error { return nil } +func (s *closeCountingStore) GetSealedInference(string, uint64) (storage.InferenceRow, bool, error) { + return storage.InferenceRow{}, false, nil +} +func (s *closeCountingStore) RecordValidationsAppliedOnce(string, []storage.ValidationObsEntry) error { + return nil +} +func (s *closeCountingStore) DrainInferenceValidationObs(string, uint64) error { return nil } +func (s *closeCountingStore) GetValidationObservability(string) ([]storage.SlotValidationObs, error) { + return nil, nil +} +func (s *closeCountingStore) Close() error { + s.closeCalls++ + return nil +} + +// TestSession_Close_ClosesUnderlyingStore proves the resource-release leg of the +// leak fix: closing a Session must close the storage it owns. The gateway-side +// tests prove rt.close() is now invoked on every automatic deactivation path; +// this proves that invocation actually frees the SQLite store the session holds. +func TestSession_Close_ClosesUnderlyingStore(t *testing.T) { + store := &closeCountingStore{} + session, _, _ := setupSessionWithOptions(t, 1, 1_000_000, 0, WithStorage(store)) + + require.NoError(t, session.Close()) + require.Equal(t, 1, store.closeCalls, "Session.Close must close the injected storage exactly once") +}