From fd808e4be01ab35d50e2fdfb08d7f3427de113a6 Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Tue, 3 Nov 2026 08:12:21 +0000 Subject: [PATCH] fix(studio): reap only stale turns when the database is shared Without a distributed setup, the chat reaper ended every claimed turn in the database at startup. When a second process shares that database, this stopped the live turns of the other process: the last outcome became interrupted, queued messages were re-queued and goals paused. Reap only the turns that stopped beating for the turn timeout, at startup and on a timer. A process that died leaves a claim that goes stale, and the next tick frees it. Assisted-by: Claude Code:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto --- core/application/studio_service.go | 40 +++++++++++------- core/application/studio_service_test.go | 56 ++++++++++++++++++------- 2 files changed, 67 insertions(+), 29 deletions(-) diff --git a/core/application/studio_service.go b/core/application/studio_service.go index 27f74fd3af..a716e036c7 100644 --- a/core/application/studio_service.go +++ b/core/application/studio_service.go @@ -228,31 +228,43 @@ const ( ) // startStudioReaper frees the chats whose turn was left running by a process -// that died. A standalone process runs once at startup with no grace: every -// claim then belongs to a process that no longer exists. In a cluster the other -// frontends are alive, so only turns that stopped beating are reaped, by the one -// frontend that holds the advisory lock at each tick. +// that died. Only turns that stopped beating are reaped: the database may be +// shared with another live process, whose turns must go on. A process that +// died leaves a claim that goes stale within studioTurnTimeout, and the next +// tick frees it. In a cluster one frontend does this at each tick, the one that +// holds the advisory lock. func (a *Application) startStudioReaper(svc *studio.Service) { ctx := a.backgroundContext() - reap := func(olderThan time.Duration) { - n, err := svc.ReapStale(ctx, olderThan) + reap := func() { + n, err := svc.ReapStale(ctx, studioTurnTimeout) if err != nil { xlog.Warn("Chats: could not reap stale turns", "error", err) } else if n > 0 { xlog.Info("Chats: ended turns whose frontend stopped", "turns", n) } } - if a.Distributed() == nil { - if a.authDB != nil { - // The JSON store settles its own claims when it loads; a database does not. - reap(0) - } - return - } if a.authDB == nil { + // The JSON store settles its own claims when it loads. return } - go advisorylock.RunLeaderLoop(ctx, a.authDB, advisorylock.KeyChatsReaper, studioReapEvery, func() { reap(studioTurnTimeout) }) + if a.Distributed() != nil { + go advisorylock.RunLeaderLoop(ctx, a.authDB, advisorylock.KeyChatsReaper, studioReapEvery, reap) + return + } + // The conditional update of a reap is safe to race with another process. + reap() + go func() { + tick := time.NewTicker(studioReapEvery) + defer tick.Stop() + for { + select { + case <-ctx.Done(): + return + case <-tick.C: + reap() + } + } + }() } // studioUsers returns the user lookup, or nil when auth is off. diff --git a/core/application/studio_service_test.go b/core/application/studio_service_test.go index fc3890a9db..a4fd709495 100644 --- a/core/application/studio_service_test.go +++ b/core/application/studio_service_test.go @@ -40,22 +40,48 @@ var _ = Describe("studio service at startup", func() { Expect(withAuth.StudioService()).NotTo(BeNil()) }) - It("frees a turn that a previous process left running", func() { - dbPath := filepath.Join(GinkgoT().TempDir(), "auth.db") - dataDir := GinkgoT().TempDir() - ctx := context.Background() - first := newApp(config.WithAuthEnabled(true), config.WithAuthDatabaseURL(dbPath), config.WithDataPath(dataDir)) - rec, err := first.StudioService().Create(ctx, "u1", studio.CreateInput{Model: "m"}) - Expect(err).NotTo(HaveOccurred()) - // The process dies in the middle of a turn: its claim is never released. - Expect(first.StudioStore().ClaimTurn(ctx, "u1", rec.ID, "dead-run")).To(Succeed()) - Expect(first.Shutdown()).To(Succeed()) + Describe("turns left running in a shared database", func() { + start := func(dbPath, dataDir string) *Application { + return newApp(config.WithAuthEnabled(true), config.WithAuthDatabaseURL(dbPath), config.WithDataPath(dataDir)) + } - second := newApp(config.WithAuthEnabled(true), config.WithAuthDatabaseURL(dbPath), config.WithDataPath(dataDir)) - info, err := second.StudioService().Turn(ctx, "u1", rec.ID) - Expect(err).NotTo(HaveOccurred()) - Expect(info.Running).To(BeFalse()) - Expect(info.Last).To(Equal(store.TurnInterrupted)) + It("keeps the turn that another live process is running", func() { + dbPath := filepath.Join(GinkgoT().TempDir(), "auth.db") + dataDir := GinkgoT().TempDir() + ctx := context.Background() + first := start(dbPath, dataDir) + rec, err := first.StudioService().Create(ctx, "u1", studio.CreateInput{Model: "m"}) + Expect(err).NotTo(HaveOccurred()) + // A second process on the same database is beating this claim. + Expect(first.StudioStore().ClaimTurn(ctx, "u1", rec.ID, "live-run")).To(Succeed()) + + second := start(dbPath, dataDir) + info, err := second.StudioService().Turn(ctx, "u1", rec.ID) + Expect(err).NotTo(HaveOccurred()) + Expect(info.Running).To(BeTrue(), "a process starting must not end a live turn of another process") + Expect(info.RunID).To(Equal("live-run")) + }) + + It("frees a turn that stopped beating when a process starts", func() { + dbPath := filepath.Join(GinkgoT().TempDir(), "auth.db") + dataDir := GinkgoT().TempDir() + ctx := context.Background() + first := start(dbPath, dataDir) + rec, err := first.StudioService().Create(ctx, "u1", studio.CreateInput{Model: "m"}) + Expect(err).NotTo(HaveOccurred()) + // The process dies in the middle of a turn: its claim is never released + // and its beat goes stale. + Expect(first.StudioStore().ClaimTurn(ctx, "u1", rec.ID, "dead-run")).To(Succeed()) + Expect(first.Shutdown()).To(Succeed()) + stale := time.Now().Add(-2 * studioTurnTimeout).UTC() + Expect(first.authDB.Exec("UPDATE studio_chats SET run_beat = ? WHERE id = ?", stale, rec.ID).Error).To(Succeed()) + + second := start(dbPath, dataDir) + info, err := second.StudioService().Turn(ctx, "u1", rec.ID) + Expect(err).NotTo(HaveOccurred()) + Expect(info.Running).To(BeFalse()) + Expect(info.Last).To(Equal(store.TurnInterrupted)) + }) }) It("gives each model its configured context size", func() {