From 6ec1fd3f4997583458dfbe0d7024c7f8cffa5b25 Mon Sep 17 00:00:00 2001 From: Dmytro Nikolayev Date: Sat, 11 Jul 2026 17:48:33 +0200 Subject: [PATCH 1/2] fix(relay): harden live config reload --- CHANGELOG.md | 6 + internal/app/app_test.go | 56 ++++ internal/app/daemon.go | 169 +++++++++-- internal/app/daemon_test.go | 420 +++++++++++++++++++++++++++- internal/app/setup.go | 7 + internal/config/config.go | 50 +++- internal/config/config_test.go | 74 +++++ internal/config/filelock.go | 55 ++++ internal/config/filelock_unix.go | 35 +++ internal/config/filelock_windows.go | 39 +++ internal/router/router.go | 190 ++++++++++++- internal/router/router_test.go | 158 ++++++++++- 12 files changed, 1203 insertions(+), 56 deletions(-) create mode 100644 internal/config/filelock.go create mode 100644 internal/config/filelock_unix.go create mode 100644 internal/config/filelock_windows.go diff --git a/CHANGELOG.md b/CHANGELOG.md index b50a28b..e757d6e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,12 @@ protocols. a config reload instead of invoking a captured stale binding. - Persistent runner cache entries are scoped to their config snapshot and are retired asynchronously, preventing stale session reuse and long reload stalls. +- Reload publication now drains the previous routing generation and performs a + final inbox reconciliation, preventing a late old-session write from being + stranded after a group or session rebind. +- Config saves now use a cross-process mutation lock, stale-revision rejection, + and atomic synced replacement so CLI changes and daemon lifecycle archives do + not silently overwrite each other or expose partial TOML to the live reloader. - Active group re-enablement now defaults the session id to its lane alias and clears a pinned runner when its session no longer matches, keeping parallel MRF lanes isolated. diff --git a/internal/app/app_test.go b/internal/app/app_test.go index 8d2fdef..0f8cf2b 100644 --- a/internal/app/app_test.go +++ b/internal/app/app_test.go @@ -6,6 +6,7 @@ import ( "context" "database/sql" "encoding/json" + "errors" "io" "os" "path/filepath" @@ -413,6 +414,61 @@ func TestSetupWizardConfiguresActiveSessionWithConfirmedAuthorizedNumber(t *test } } +type setupCreateGroupHookTransport struct { + transport.ChatTransport + hook func() error +} + +func (t *setupCreateGroupHookTransport) CreateGroup(ctx context.Context, name string, participants []string) (*types.Chat, error) { + if err := t.hook(); err != nil { + return nil, err + } + return t.ChatTransport.CreateGroup(ctx, name, participants) +} + +func TestSetupWizardRejectsConcurrentUpdateAfterCreatingConfig(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + path := filepath.Join(t.TempDir(), "config.toml") + baseTransport := fake.New(nil) + hookedTransport := &setupCreateGroupHookTransport{ + ChatTransport: baseTransport, + hook: func() error { + concurrent, err := config.Load(path) + if err != nil { + return err + } + concurrent.App.LogLevel = "debug" + return config.Save(path, concurrent) + }, + } + state := &cliState{ + configPath: path, + transportFactory: func(context.Context, config.Config) (transport.ChatTransport, error) { + return hookedTransport, nil + }, + } + cmd := state.setupCommand() + cmd.SetArgs([]string{ + "--yes", + "--agent", "codex", + "--authorized", "+1 (555) 000-1111", + "--group-name", "Coderoam Test", + "--workdir", t.TempDir(), + "--session-id", "codex-session", + }) + _, err := captureStdout(t, cmd.Execute) + if !errors.Is(err, config.ErrConfigChanged) { + t.Fatalf("setup error = %v, want stale config rejection", err) + } + loaded, loadErr := config.Load(path) + if loadErr != nil { + t.Fatal(loadErr) + } + if loaded.App.LogLevel != "debug" { + t.Fatalf("concurrent update was overwritten: log_level=%q", loaded.App.LogLevel) + } +} + func TestSetupWizardDefaultsGroupNameFromSelectedAgent(t *testing.T) { t.Setenv("HOME", t.TempDir()) cfg := config.Default() diff --git a/internal/app/daemon.go b/internal/app/daemon.go index 33c4849..7be7af2 100644 --- a/internal/app/daemon.go +++ b/internal/app/daemon.go @@ -2,6 +2,7 @@ package app import ( "context" + "errors" "fmt" "maps" "os" @@ -306,21 +307,22 @@ type runIncomingHandler interface { } type runConfigTarget interface { - SetConfig(config.Config) + SetConfigAndWait(context.Context, config.Config) error ScheduleUnreadActiveFallbacks(context.Context, *config.Config, int) (int, error) } type runConfigManager struct { - mu sync.Mutex - path string - profileOverride string - holder *runConfigHolder - target runConfigTarget - store *db.Store - transport transport.ChatTransport - logf func(string, ...any) - reconcilePending bool - pendingLifecycle []types.GroupEvent + mu sync.Mutex + path string + profileOverride string + holder *runConfigHolder + target runConfigTarget + store *db.Store + transport transport.ChatTransport + logf func(string, ...any) + reconcilePending bool + generationDrainPending bool + pendingLifecycle []types.GroupEvent } type runConfigRefreshingHandler struct { @@ -346,6 +348,9 @@ func (m *runConfigManager) Refresh(ctx context.Context) (bool, error) { defer m.mu.Unlock() changed, err := m.refreshLocked(ctx) if err != nil { + if retryErr := m.retryPendingLifecycleAgainstDiskLocked(ctx); retryErr != nil { + return changed, errors.Join(err, retryErr) + } return changed, err } if err := m.retryPendingLifecycleLocked(ctx); err != nil { @@ -364,7 +369,7 @@ func (m *runConfigManager) refreshLocked(ctx context.Context) (bool, error) { } current := m.holder.Load() changed := !reflect.DeepEqual(current, updated) - if !changed && !m.reconcilePending { + if !changed && !m.reconcilePending && !m.generationDrainPending { return false, nil } if changed { @@ -373,14 +378,31 @@ func (m *runConfigManager) refreshLocked(ctx context.Context) (bool, error) { } } - migrated, repaired, err := reconcileRunConfigBindings(ctx, m.store, updated) - if err != nil { - return false, err - } + migrated := 0 + repaired := 0 if changed { + preMigrated, preRepaired, err := reconcileRunConfigBindings(ctx, m.store, updated) + if err != nil { + return false, err + } + migrated += preMigrated + repaired += preRepaired m.holder.Store(updated) - m.target.SetConfig(updated) + m.reconcilePending = true + m.generationDrainPending = true } + if m.generationDrainPending { + if err := m.target.SetConfigAndWait(ctx, updated); err != nil { + return changed, err + } + m.generationDrainPending = false + } + postMigrated, postRepaired, err := reconcileRunConfigBindings(ctx, m.store, updated) + if err != nil { + return changed, err + } + migrated += postMigrated + repaired += postRepaired m.reconcilePending = true rescheduled, err := m.target.ScheduleUnreadActiveFallbacks(ctx, &updated, 100) if err != nil { @@ -414,12 +436,19 @@ func reconcileRunConfigBindings(ctx context.Context, store *db.Store, cfg config } func (m *runConfigManager) HandleRelayGroupLifecycleEvent(ctx context.Context, event types.GroupEvent) (bool, error) { + if archive, _ := shouldArchiveRelayGroup(event); !archive { + return false, nil + } m.mu.Lock() defer m.mu.Unlock() if _, err := m.refreshLocked(ctx); err != nil { + archived, diskErr := m.applyLifecycleEventToLatestDiskLocked(ctx, event) + if diskErr == nil { + return archived, nil + } m.queueLifecycleEventLocked(event) - return false, fmt.Errorf("refresh config before group lifecycle event; event queued for retry: %w", err) + return false, fmt.Errorf("refresh config before group lifecycle event: %w; disk lifecycle update failed and event was queued: %w", err, diskErr) } if err := m.retryPendingLifecycleLocked(ctx); err != nil { m.queueLifecycleEventLocked(event) @@ -432,14 +461,78 @@ func (m *runConfigManager) HandleRelayGroupLifecycleEvent(ctx context.Context, e return archived, err } +func (m *runConfigManager) applyLifecycleEventToLatestDiskLocked(ctx context.Context, event types.GroupEvent) (bool, error) { + const maxConfigSaveAttempts = 5 + var updated config.Config + var archived bool + var err error + for attempt := 1; attempt <= maxConfigSaveAttempts; attempt++ { + diskCfg, loadErr := config.Load(m.path) + if loadErr != nil { + return false, loadErr + } + updated, archived, err = handleRelayGroupLifecycleEventForProfile(ctx, diskCfg, m.path, m.store, m.transport, event, m.holder.Load().App.Profile) + if err == nil { + if !archived && relayGroupArchived(diskCfg, event.ChatID) { + updated = diskCfg + archived = true + } + break + } + if !errors.Is(err, config.ErrConfigChanged) { + return false, err + } + } + if err != nil { + return false, fmt.Errorf("archive relay group after %d concurrent config updates: %w", maxConfigSaveAttempts, err) + } + if !archived { + return false, nil + } + liveCfg := m.holder.Load() + for _, diskGroup := range updated.Groups { + if diskGroup.ID != event.ChatID { + continue + } + for i, liveGroup := range liveCfg.Groups { + if liveGroup.ID != event.ChatID { + continue + } + liveCfg.Groups = slices.Clone(liveCfg.Groups) + liveCfg.Groups[i].Enabled = false + liveCfg.Groups[i].Archived = true + liveCfg.Groups[i].ArchivedAt = diskGroup.ArchivedAt + liveCfg.Groups[i].ArchiveReason = diskGroup.ArchiveReason + return true, m.publishLifecycleConfigLocked(ctx, liveCfg, event.ChatID) + } + break + } + return true, nil +} + func (m *runConfigManager) applyLifecycleEventLocked(ctx context.Context, event types.GroupEvent) (bool, error) { - updated, archived, err := handleRelayGroupLifecycleEvent(ctx, m.holder.Load(), m.path, m.store, m.transport, event) + liveCfg := m.holder.Load() + if relayGroupArchived(liveCfg, event.ChatID) { + return true, m.publishLifecycleConfigLocked(ctx, liveCfg, event.ChatID) + } + updated, archived, err := handleRelayGroupLifecycleEvent(ctx, liveCfg, m.path, m.store, m.transport, event) if err != nil || !archived { return archived, err } - m.holder.Store(updated) - m.target.SetConfig(updated) - return true, nil + return true, m.publishLifecycleConfigLocked(ctx, updated, event.ChatID) +} + +func (m *runConfigManager) publishLifecycleConfigLocked(ctx context.Context, cfg config.Config, chatID string) error { + m.holder.Store(cfg) + m.generationDrainPending = true + if err := m.target.SetConfigAndWait(ctx, cfg); err != nil { + return err + } + m.generationDrainPending = false + // A previous-generation handler may have written after the lifecycle + // helper's first cleanup but before the config generation drained. + _, err := m.store.DeleteChatData(ctx, cfg.App.Profile, chatID) + return err } func (m *runConfigManager) queueLifecycleEventLocked(event types.GroupEvent) { @@ -466,6 +559,30 @@ func (m *runConfigManager) retryPendingLifecycleLocked(ctx context.Context) erro return nil } +func (m *runConfigManager) retryPendingLifecycleAgainstDiskLocked(ctx context.Context) error { + for len(m.pendingLifecycle) > 0 { + event := m.pendingLifecycle[0] + archived, err := m.applyLifecycleEventToLatestDiskLocked(ctx, event) + if err != nil { + return fmt.Errorf("retry queued group lifecycle event against disk config: %w", err) + } + m.pendingLifecycle = m.pendingLifecycle[1:] + if archived && m.logf != nil { + m.logf("[group-event] chat=%s archived=true reason=queued-disk-retry\n", logging.Redact(event.ChatID)) + } + } + return nil +} + +func relayGroupArchived(cfg config.Config, chatID string) bool { + for _, group := range cfg.Groups { + if group.ID == chatID && group.Mode == config.GroupModeActiveSession && group.RelayManaged && group.Archived && !group.Enabled { + return true + } + } + return false +} + func validateRunConfigReload(current config.Config, updated config.Config) error { if current.App != updated.App { return fmt.Errorf("app settings changed; restart coderoam to apply them") @@ -879,6 +996,10 @@ func activeReadReceiptBatches(records []db.ActiveReadReceiptRecord) [][]db.Activ } func handleRelayGroupLifecycleEvent(ctx context.Context, cfg config.Config, configPath string, store *db.Store, chatTransport transport.ChatTransport, event types.GroupEvent) (config.Config, bool, error) { + return handleRelayGroupLifecycleEventForProfile(ctx, cfg, configPath, store, chatTransport, event, cfg.App.Profile) +} + +func handleRelayGroupLifecycleEventForProfile(ctx context.Context, cfg config.Config, configPath string, store *db.Store, chatTransport transport.ChatTransport, event types.GroupEvent, profileID string) (config.Config, bool, error) { groupIndex := -1 for i, group := range cfg.Groups { if group.ID == event.ChatID && group.Mode == config.GroupModeActiveSession && group.RelayManaged && group.Enabled && !group.Archived { @@ -900,7 +1021,7 @@ func handleRelayGroupLifecycleEvent(ctx context.Context, cfg config.Config, conf archiveErrText = err.Error() } } - deletedRows, err := store.DeleteChatData(ctx, cfg.App.Profile, group.ID) + deletedRows, err := store.DeleteChatData(ctx, profileID, group.ID) if err != nil { return cfg, false, err } @@ -916,7 +1037,7 @@ func handleRelayGroupLifecycleEvent(ctx context.Context, cfg config.Config, conf if err := config.Save(configPath, cfg); err != nil { return cfg, false, err } - _ = store.Audit(ctx, cfg.App.Profile, "relay_group_archived", event.SenderID, group.ID, map[string]any{ + _ = store.Audit(ctx, profileID, "relay_group_archived", event.SenderID, group.ID, map[string]any{ "alias": group.Alias, "session_id": config.ActiveSessionID(group), "reason": reason, diff --git a/internal/app/daemon_test.go b/internal/app/daemon_test.go index f575b9c..ab0bc8e 100644 --- a/internal/app/daemon_test.go +++ b/internal/app/daemon_test.go @@ -5,6 +5,7 @@ import ( "fmt" "os" "path/filepath" + "slices" "strings" "sync" "testing" @@ -481,6 +482,270 @@ func TestRunConfigManagerRetriesLifecycleEventAfterInvalidConfig(t *testing.T) { } } +func TestRunConfigManagerDoesNotLetBenignEventHideQueuedArchive(t *testing.T) { + t.Parallel() + ctx := t.Context() + dir := t.TempDir() + path := filepath.Join(dir, "config.toml") + initial := config.Default() + initial.App.Profile = "test" + initial.App.DatabasePath = filepath.Join(dir, "coderoam.sqlite3") + initial.Transport.Type = "fake" + initial.Groups = []config.GroupConfig{{ + ID: "mrf-1@g.us", + Alias: "mrf-1", + Mode: config.GroupModeActiveSession, + ActiveSessionID: "mrf-1", + Enabled: true, + RelayManaged: true, + }} + if err := config.Save(path, initial); err != nil { + t.Fatal(err) + } + store, err := db.Open(initial.App.DatabasePath) + if err != nil { + t.Fatal(err) + } + defer store.Close() + if err := store.EnsureProfile(ctx, initial.App.Profile); err != nil { + t.Fatal(err) + } + bridgeRouter := router.New(initial, store, nil) + defer bridgeRouter.Stop(context.Background()) + holder := newRunConfigHolder(initial) + manager := newRunConfigManager(path, "", holder, bridgeRouter, store, nil, nil) + if err := os.WriteFile(path, []byte("[[groups]\ninvalid"), 0o600); err != nil { + t.Fatal(err) + } + + archived, err := manager.HandleRelayGroupLifecycleEvent(ctx, types.GroupEvent{ + ChatID: "mrf-1@g.us", + ParticipantCount: 3, + Timestamp: time.Now(), + }) + if err != nil || archived || len(manager.pendingLifecycle) != 0 { + t.Fatalf("benign event archived=%t err=%v pending=%+v", archived, err, manager.pendingLifecycle) + } + archived, err = manager.HandleRelayGroupLifecycleEvent(ctx, types.GroupEvent{ + ChatID: "mrf-1@g.us", + Deleted: true, + Timestamp: time.Now(), + }) + if err == nil || archived || len(manager.pendingLifecycle) != 1 || !manager.pendingLifecycle[0].Deleted { + t.Fatalf("delete event archived=%t err=%v pending=%+v", archived, err, manager.pendingLifecycle) + } + if err := config.Save(path, initial); err != nil { + t.Fatal(err) + } + if _, err := manager.Refresh(ctx); err != nil { + t.Fatal(err) + } + loaded, err := config.Load(path) + if err != nil { + t.Fatal(err) + } + if len(loaded.Groups) != 1 || !loaded.Groups[0].Archived { + t.Fatalf("queued delete was not applied: %+v", loaded.Groups) + } +} + +func TestRunConfigManagerArchivesThroughRestartOnlyDiskChange(t *testing.T) { + t.Parallel() + ctx := t.Context() + dir := t.TempDir() + path := filepath.Join(dir, "config.toml") + initial := config.Default() + initial.App.Profile = "test" + initial.App.DatabasePath = filepath.Join(dir, "coderoam.sqlite3") + initial.Transport.Type = "fake" + initial.Groups = []config.GroupConfig{{ + ID: "mrf-1@g.us", + Alias: "mrf-1", + Mode: config.GroupModeActiveSession, + ActiveSessionID: "mrf-1", + Enabled: true, + RelayManaged: true, + }} + if err := config.Save(path, initial); err != nil { + t.Fatal(err) + } + store, err := db.Open(initial.App.DatabasePath) + if err != nil { + t.Fatal(err) + } + defer store.Close() + if err := store.EnsureProfile(ctx, initial.App.Profile); err != nil { + t.Fatal(err) + } + bridgeRouter := router.New(initial, store, nil) + defer bridgeRouter.Stop(context.Background()) + holder := newRunConfigHolder(initial) + manager := newRunConfigManager(path, "", holder, bridgeRouter, store, nil, nil) + restartOnly, err := config.Load(path) + if err != nil { + t.Fatal(err) + } + restartOnly.Transport.DownloadMedia = true + if err := config.Save(path, restartOnly); err != nil { + t.Fatal(err) + } + + archived, err := manager.HandleRelayGroupLifecycleEvent(ctx, types.GroupEvent{ + ChatID: "mrf-1@g.us", + Deleted: true, + Timestamp: time.Now(), + }) + if err != nil || !archived { + t.Fatalf("restart-only lifecycle archived=%t err=%v", archived, err) + } + if len(manager.pendingLifecycle) != 0 { + t.Fatalf("restart-only lifecycle was left pending: %+v", manager.pendingLifecycle) + } + diskCfg, err := config.Load(path) + if err != nil { + t.Fatal(err) + } + if !diskCfg.Transport.DownloadMedia || len(diskCfg.Groups) != 1 || !diskCfg.Groups[0].Archived { + t.Fatalf("disk config lost restart-only change or archive: %+v", diskCfg) + } + liveCfg := holder.Load() + if liveCfg.Transport.DownloadMedia || len(liveCfg.Groups) != 1 || !liveCfg.Groups[0].Archived { + t.Fatalf("live config did not isolate restart-only change while archiving: %+v", liveCfg) + } +} + +func TestRunConfigManagerRetriesQueuedLifecycleThroughRestartOnlyDiskChange(t *testing.T) { + t.Parallel() + ctx := t.Context() + dir := t.TempDir() + path := filepath.Join(dir, "config.toml") + initial := config.Default() + initial.App.Profile = "test" + initial.App.DatabasePath = filepath.Join(dir, "coderoam.sqlite3") + initial.Transport.Type = "fake" + initial.Groups = []config.GroupConfig{{ + ID: "mrf-1@g.us", + Alias: "mrf-1", + Mode: config.GroupModeActiveSession, + ActiveSessionID: "mrf-1", + Enabled: true, + RelayManaged: true, + }} + if err := config.Save(path, initial); err != nil { + t.Fatal(err) + } + runtimeConfig, err := config.Load(path) + if err != nil { + t.Fatal(err) + } + store, err := db.Open(initial.App.DatabasePath) + if err != nil { + t.Fatal(err) + } + defer store.Close() + if err := store.EnsureProfile(ctx, initial.App.Profile); err != nil { + t.Fatal(err) + } + target := &retryingRunConfigTarget{} + manager := newRunConfigManager(path, "", newRunConfigHolder(runtimeConfig), target, store, nil, nil) + manager.pendingLifecycle = []types.GroupEvent{{ + ChatID: "mrf-1@g.us", + Deleted: true, + Timestamp: time.Now(), + }} + restartOnly := runtimeConfig + restartOnly.Transport.DownloadMedia = true + if err := config.Save(path, restartOnly); err != nil { + t.Fatal(err) + } + + changed, err := manager.Refresh(ctx) + if err == nil || changed { + t.Fatalf("restart-only refresh changed=%t err=%v", changed, err) + } + if len(manager.pendingLifecycle) != 0 { + t.Fatalf("queued lifecycle was not drained: %+v", manager.pendingLifecycle) + } + diskCfg, loadErr := config.Load(path) + if loadErr != nil { + t.Fatal(loadErr) + } + if !diskCfg.Transport.DownloadMedia || !relayGroupArchived(diskCfg, "mrf-1@g.us") { + t.Fatalf("disk config lost restart-only change or archive: %+v", diskCfg) + } + liveCfg := manager.holder.Load() + if liveCfg.Transport.DownloadMedia || !relayGroupArchived(liveCfg, "mrf-1@g.us") { + t.Fatalf("live config did not isolate restart-only change while archiving: %+v", liveCfg) + } +} + +func TestRunConfigManagerRetriesCleanupAfterInterruptedLifecyclePublication(t *testing.T) { + t.Parallel() + ctx := t.Context() + dir := t.TempDir() + path := filepath.Join(dir, "config.toml") + initial := config.Default() + initial.App.Profile = "test" + initial.App.DatabasePath = filepath.Join(dir, "coderoam.sqlite3") + initial.Transport.Type = "fake" + initial.Groups = []config.GroupConfig{{ + ID: "mrf-1@g.us", + Alias: "mrf-1", + Mode: config.GroupModeActiveSession, + ActiveSessionID: "mrf-1", + Enabled: true, + RelayManaged: true, + }} + if err := config.Save(path, initial); err != nil { + t.Fatal(err) + } + runtimeConfig, err := config.Load(path) + if err != nil { + t.Fatal(err) + } + store, err := db.Open(initial.App.DatabasePath) + if err != nil { + t.Fatal(err) + } + defer store.Close() + if err := store.EnsureProfile(ctx, initial.App.Profile); err != nil { + t.Fatal(err) + } + target := &interruptingRunConfigTarget{} + manager := newRunConfigManager(path, "", newRunConfigHolder(runtimeConfig), target, store, nil, nil) + archived, err := manager.HandleRelayGroupLifecycleEvent(ctx, types.GroupEvent{ + ChatID: "mrf-1@g.us", + Deleted: true, + Timestamp: time.Now(), + }) + if err == nil || !archived || len(manager.pendingLifecycle) != 1 || !manager.generationDrainPending { + t.Fatalf("interrupted lifecycle archived=%t err=%v pending=%+v drain_pending=%t", archived, err, manager.pendingLifecycle, manager.generationDrainPending) + } + if _, _, err := store.StoreActiveInboxMessage(ctx, initial.App.Profile, "mrf-1", "mrf-1", types.IncomingMessage{ + ID: "late-old-generation", + ChatID: "mrf-1@g.us", + ChatType: types.ChatTypeGroup, + SenderID: "owner@lid", + Text: "late write", + Timestamp: time.Now(), + }); err != nil { + t.Fatal(err) + } + manager.mu.Lock() + err = manager.retryPendingLifecycleLocked(ctx) + manager.mu.Unlock() + if err != nil { + t.Fatal(err) + } + rows, err := store.ListActiveInbox(ctx, initial.App.Profile, "unread", 10) + if err != nil { + t.Fatal(err) + } + if len(rows) != 0 || len(manager.pendingLifecycle) != 0 || manager.generationDrainPending || target.setCount != 2 { + t.Fatalf("cleanup retry rows=%+v pending=%+v drain_pending=%t set_count=%d", rows, manager.pendingLifecycle, manager.generationDrainPending, target.setCount) + } +} + type retryingRunConfigTarget struct { setCount int scheduleCount int @@ -488,9 +753,47 @@ type retryingRunConfigTarget struct { lastConfigured config.Config } -func (t *retryingRunConfigTarget) SetConfig(cfg config.Config) { +type interruptingRunConfigTarget struct { + setCount int + scheduleCount int +} + +func (t *interruptingRunConfigTarget) SetConfigAndWait(context.Context, config.Config) error { + t.setCount++ + if t.setCount == 1 { + return context.DeadlineExceeded + } + return nil +} + +func (t *interruptingRunConfigTarget) ScheduleUnreadActiveFallbacks(context.Context, *config.Config, int) (int, error) { + t.scheduleCount++ + return 0, nil +} + +type blockingRunConfigTarget struct { + published chan struct{} + release chan struct{} +} + +func (t *blockingRunConfigTarget) SetConfigAndWait(ctx context.Context, _ config.Config) error { + close(t.published) + select { + case <-t.release: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +func (t *blockingRunConfigTarget) ScheduleUnreadActiveFallbacks(context.Context, *config.Config, int) (int, error) { + return 0, nil +} + +func (t *retryingRunConfigTarget) SetConfigAndWait(_ context.Context, cfg config.Config) error { t.setCount++ t.lastConfigured = cfg + return nil } func (t *retryingRunConfigTarget) ScheduleUnreadActiveFallbacks(context.Context, *config.Config, int) (int, error) { @@ -547,6 +850,48 @@ func TestRunConfigManagerRetriesPostPublishReconciliation(t *testing.T) { } } +func TestRunConfigManagerResumesInterruptedGenerationDrain(t *testing.T) { + t.Parallel() + ctx := t.Context() + dir := t.TempDir() + path := filepath.Join(dir, "config.toml") + initial := config.Default() + initial.App.Profile = "test" + initial.App.DatabasePath = filepath.Join(dir, "coderoam.sqlite3") + initial.Transport.Type = "fake" + if err := config.Save(path, initial); err != nil { + t.Fatal(err) + } + runtimeConfig, err := config.Load(path) + if err != nil { + t.Fatal(err) + } + store, err := db.Open(initial.App.DatabasePath) + if err != nil { + t.Fatal(err) + } + defer store.Close() + if err := store.EnsureProfile(ctx, initial.App.Profile); err != nil { + t.Fatal(err) + } + target := &interruptingRunConfigTarget{} + manager := newRunConfigManager(path, "", newRunConfigHolder(runtimeConfig), target, store, nil, nil) + updated := runtimeConfig + updated.Active.AckMode = "verbose" + if err := config.Save(path, updated); err != nil { + t.Fatal(err) + } + + changed, err := manager.Refresh(ctx) + if err == nil || !changed || !manager.generationDrainPending || target.setCount != 1 { + t.Fatalf("interrupted refresh changed=%t err=%v drain_pending=%t set_count=%d", changed, err, manager.generationDrainPending, target.setCount) + } + changed, err = manager.Refresh(ctx) + if err != nil || changed || manager.generationDrainPending || manager.reconcilePending || target.setCount != 2 || target.scheduleCount != 1 { + t.Fatalf("resumed refresh changed=%t err=%v drain_pending=%t reconcile_pending=%t sets=%d schedules=%d", changed, err, manager.generationDrainPending, manager.reconcilePending, target.setCount, target.scheduleCount) + } +} + func TestRunConfigManagerDoesNotPublishBeforeBindingReconciliation(t *testing.T) { t.Parallel() ctx := t.Context() @@ -593,6 +938,79 @@ func TestRunConfigManagerDoesNotPublishBeforeBindingReconciliation(t *testing.T) } } +func TestRunConfigManagerRepairsWritesFromDrainedGeneration(t *testing.T) { + t.Parallel() + ctx := t.Context() + dir := t.TempDir() + path := filepath.Join(dir, "config.toml") + initial := config.Default() + initial.App.Profile = "test" + initial.App.DatabasePath = filepath.Join(dir, "coderoam.sqlite3") + initial.Transport.Type = "fake" + initial.Groups = []config.GroupConfig{{ + ID: "mrf-1@g.us", + Alias: "mrf-1", + Mode: config.GroupModeActiveSession, + ActiveSessionID: "old-session", + Enabled: true, + }} + if err := config.Save(path, initial); err != nil { + t.Fatal(err) + } + runtimeConfig, err := config.Load(path) + if err != nil { + t.Fatal(err) + } + store, err := db.Open(initial.App.DatabasePath) + if err != nil { + t.Fatal(err) + } + defer store.Close() + if err := store.EnsureProfile(ctx, initial.App.Profile); err != nil { + t.Fatal(err) + } + target := &blockingRunConfigTarget{published: make(chan struct{}), release: make(chan struct{})} + holder := newRunConfigHolder(runtimeConfig) + manager := newRunConfigManager(path, "", holder, target, store, nil, nil) + updated := runtimeConfig + updated.Groups = slices.Clone(runtimeConfig.Groups) + updated.Groups[0].ActiveSessionID = "new-session" + if err := config.Save(path, updated); err != nil { + t.Fatal(err) + } + refreshDone := make(chan error, 1) + go func() { + _, err := manager.Refresh(ctx) + refreshDone <- err + }() + select { + case <-target.published: + case <-time.After(2 * time.Second): + t.Fatal("updated config generation was not published") + } + if _, _, err := store.StoreActiveInboxMessage(ctx, initial.App.Profile, "mrf-1", "old-session", types.IncomingMessage{ + ID: "late-old-generation", + ChatID: "mrf-1@g.us", + ChatType: types.ChatTypeGroup, + SenderID: "owner@lid", + Text: "late write", + Timestamp: time.Now(), + }); err != nil { + t.Fatal(err) + } + close(target.release) + if err := <-refreshDone; err != nil { + t.Fatal(err) + } + rows, err := store.ListActiveInbox(ctx, initial.App.Profile, "unread", 10) + if err != nil { + t.Fatal(err) + } + if len(rows) != 1 || rows[0].SessionID != "new-session" { + t.Fatalf("late previous-generation row was not repaired: %+v", rows) + } +} + func TestRunConfigManagerAppliesRuntimeProfileOverride(t *testing.T) { t.Parallel() ctx := t.Context() diff --git a/internal/app/setup.go b/internal/app/setup.go index 22fe905..3001d1e 100644 --- a/internal/app/setup.go +++ b/internal/app/setup.go @@ -148,6 +148,13 @@ func (s *cliState) runSetupWizard(cmd *cobra.Command, opts setupWizardOptions) e if err := config.Save(path, cfg); err != nil { return err } + // Load the just-created file so the final save is guarded by its exact + // revision. Setup can wait a long time for login or group creation; a + // zero-revision config would otherwise overwrite concurrent updates. + cfg, err = config.Load(path) + if err != nil { + return err + } } workdir := strings.TrimSpace(opts.Workdir) diff --git a/internal/config/config.go b/internal/config/config.go index 433f3b6..7e64ceb 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -1,6 +1,7 @@ package config import ( + "crypto/sha256" "errors" "fmt" "os" @@ -20,19 +21,23 @@ const ( ) type Config struct { - App AppConfig `toml:"app"` - Transport TransportConfig `toml:"transport"` - Trigger TriggerConfig `toml:"trigger"` - Active ActiveConfig `toml:"active"` - Security SecurityConfig `toml:"security"` - RateLimits RateLimitConfig `toml:"rate_limits"` - Reply ReplyConfig `toml:"reply"` - Session SessionConfig `toml:"session"` - Retention RetentionConfig `toml:"retention"` - Concurrency ConcurrencyConfig `toml:"concurrency"` - Runner map[string]RunnerConfig `toml:"runner"` - Groups []GroupConfig `toml:"groups"` -} + App AppConfig `toml:"app"` + Transport TransportConfig `toml:"transport"` + Trigger TriggerConfig `toml:"trigger"` + Active ActiveConfig `toml:"active"` + Security SecurityConfig `toml:"security"` + RateLimits RateLimitConfig `toml:"rate_limits"` + Reply ReplyConfig `toml:"reply"` + Session SessionConfig `toml:"session"` + Retention RetentionConfig `toml:"retention"` + Concurrency ConcurrencyConfig `toml:"concurrency"` + Runner map[string]RunnerConfig `toml:"runner"` + Groups []GroupConfig `toml:"groups"` + sourcePath string + sourceRevision [sha256.Size]byte +} + +var ErrConfigChanged = errors.New("config changed since it was loaded") type AppConfig struct { Profile string `toml:"profile"` @@ -235,6 +240,8 @@ func Load(path string) (Config, error) { if err := ValidateActiveSessionBindings(cfg); err != nil { return Config{}, err } + cfg.sourcePath = filepath.Clean(path) + cfg.sourceRevision = sha256.Sum256(data) return cfg, nil } @@ -268,7 +275,22 @@ func Save(path string, cfg Config) error { if err != nil { return err } - return os.WriteFile(path, data, 0o600) + return saveConfigFile(path, data, cfg) +} + +func saveConfigFile(path string, data []byte, cfg Config) error { + return withConfigMutationLock(path, func() error { + if cfg.sourcePath == filepath.Clean(path) && cfg.sourceRevision != ([sha256.Size]byte{}) { + current, err := os.ReadFile(path) + if err != nil { + return err + } + if sha256.Sum256(current) != cfg.sourceRevision { + return fmt.Errorf("%w at %s; reload and retry", ErrConfigChanged, path) + } + } + return atomicWriteConfig(path, data) + }) } func ApplyDefaults(cfg *Config) { diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 8818de9..8235561 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -1,6 +1,7 @@ package config import ( + "errors" "os" "path/filepath" "strings" @@ -205,3 +206,76 @@ func TestSaveRejectsDuplicateActiveSessionBindings(t *testing.T) { t.Fatalf("error = %v, want duplicate session save guard", err) } } + +func TestSaveRejectsStaleLoadedConfig(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "config.toml") + if err := Save(path, Default()); err != nil { + t.Fatal(err) + } + first, err := Load(path) + if err != nil { + t.Fatal(err) + } + stale, err := Load(path) + if err != nil { + t.Fatal(err) + } + first.Active.AckMode = "verbose" + if err := Save(path, first); err != nil { + t.Fatal(err) + } + stale.Active.AckMode = "off" + if err := Save(path, stale); !errors.Is(err, ErrConfigChanged) { + t.Fatalf("stale Save error = %v, want ErrConfigChanged", err) + } + loaded, err := Load(path) + if err != nil { + t.Fatal(err) + } + if loaded.Active.AckMode != "verbose" { + t.Fatalf("stale save replaced newer config: ack_mode=%q", loaded.Active.AckMode) + } +} + +func TestSaveNeverExposesPartialConfig(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "config.toml") + if err := Save(path, Default()); err != nil { + t.Fatal(err) + } + stop := make(chan struct{}) + readerDone := make(chan error, 1) + go func() { + for { + select { + case <-stop: + readerDone <- nil + return + default: + } + if _, err := Load(path); err != nil { + readerDone <- err + return + } + } + }() + for i := 0; i < 50; i++ { + cfg, err := Load(path) + if err != nil { + t.Fatal(err) + } + if i%2 == 0 { + cfg.Active.AckMode = "verbose" + } else { + cfg.Active.AckMode = "off" + } + if err := Save(path, cfg); err != nil { + t.Fatal(err) + } + } + close(stop) + if err := <-readerDone; err != nil { + t.Fatalf("concurrent config read observed partial write: %v", err) + } +} diff --git a/internal/config/filelock.go b/internal/config/filelock.go new file mode 100644 index 0000000..3116891 --- /dev/null +++ b/internal/config/filelock.go @@ -0,0 +1,55 @@ +package config + +import ( + "errors" + "os" + "path/filepath" +) + +func withConfigMutationLock(path string, mutate func() error) error { + dir := filepath.Dir(path) + if err := os.MkdirAll(dir, 0o700); err != nil { + return err + } + lock, err := os.OpenFile(path+".lock", os.O_CREATE|os.O_RDWR, 0o600) + if err != nil { + return err + } + defer lock.Close() + if err := lockConfigFile(lock); err != nil { + return err + } + defer func() { _ = unlockConfigFile(lock) }() + return mutate() +} + +func atomicWriteConfig(path string, data []byte) (err error) { + dir := filepath.Dir(path) + tmp, err := os.CreateTemp(dir, "."+filepath.Base(path)+".tmp-*") + if err != nil { + return err + } + tmpPath := tmp.Name() + defer func() { + _ = tmp.Close() + if removeErr := os.Remove(tmpPath); removeErr != nil && !errors.Is(removeErr, os.ErrNotExist) && err == nil { + err = removeErr + } + }() + if err = tmp.Chmod(0o600); err != nil { + return err + } + if _, err = tmp.Write(data); err != nil { + return err + } + if err = tmp.Sync(); err != nil { + return err + } + if err = tmp.Close(); err != nil { + return err + } + if err = replaceConfigFile(tmpPath, path); err != nil { + return err + } + return syncConfigDirectory(dir) +} diff --git a/internal/config/filelock_unix.go b/internal/config/filelock_unix.go new file mode 100644 index 0000000..734e532 --- /dev/null +++ b/internal/config/filelock_unix.go @@ -0,0 +1,35 @@ +//go:build !windows + +package config + +import ( + "os" + "syscall" +) + +func lockConfigFile(file *os.File) error { + for { + err := syscall.Flock(int(file.Fd()), syscall.LOCK_EX) + if err == syscall.EINTR { + continue + } + return err + } +} + +func unlockConfigFile(file *os.File) error { + return syscall.Flock(int(file.Fd()), syscall.LOCK_UN) +} + +func replaceConfigFile(from string, to string) error { + return os.Rename(from, to) +} + +func syncConfigDirectory(path string) error { + dir, err := os.Open(path) + if err != nil { + return err + } + defer dir.Close() + return dir.Sync() +} diff --git a/internal/config/filelock_windows.go b/internal/config/filelock_windows.go new file mode 100644 index 0000000..d971442 --- /dev/null +++ b/internal/config/filelock_windows.go @@ -0,0 +1,39 @@ +//go:build windows + +package config + +import ( + "os" + + "golang.org/x/sys/windows" +) + +const configLockRangeOffsetHigh = 0x7FFFFFFE + +func configLockOverlapped() *windows.Overlapped { + return &windows.Overlapped{Offset: 0, OffsetHigh: configLockRangeOffsetHigh} +} + +func lockConfigFile(file *os.File) error { + return windows.LockFileEx(windows.Handle(file.Fd()), windows.LOCKFILE_EXCLUSIVE_LOCK, 0, 1, 0, configLockOverlapped()) +} + +func unlockConfigFile(file *os.File) error { + return windows.UnlockFileEx(windows.Handle(file.Fd()), 0, 1, 0, configLockOverlapped()) +} + +func replaceConfigFile(from string, to string) error { + fromPath, err := windows.UTF16PtrFromString(from) + if err != nil { + return err + } + toPath, err := windows.UTF16PtrFromString(to) + if err != nil { + return err + } + return windows.MoveFileEx(fromPath, toPath, windows.MOVEFILE_REPLACE_EXISTING|windows.MOVEFILE_WRITE_THROUGH) +} + +func syncConfigDirectory(string) error { + return nil +} diff --git a/internal/router/router.go b/internal/router/router.go index ffebd8e..a88298a 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -34,14 +34,21 @@ type Router struct { activeFallbackDelay time.Duration activeFallbackLimit int activeFallbackScheduled map[string]bool + configUses map[*config.Config]*routerConfigUse // stopped is closed exactly once by Stop; scheduled fallback goroutines // watch it so shutdown can drain them deterministically. stopped chan struct{} stopOnce sync.Once + stopping bool fallbackWG sync.WaitGroup retiredWG sync.WaitGroup } +type routerConfigUse struct { + count int + done chan struct{} +} + type ProcessResult struct { Ignored bool Reason string @@ -62,6 +69,7 @@ func New(cfg config.Config, store *db.Store, chatTransport transport.ChatTranspo activeFallbackDelay: time.Duration(cfg.Active.FallbackDelaySeconds) * time.Second, activeFallbackLimit: cfg.Active.FallbackBatchLimit, activeFallbackScheduled: map[string]bool{}, + configUses: map[*config.Config]*routerConfigUse{}, stopped: make(chan struct{}), } r.cfg.Store(snapshotConfig(cfg)) @@ -91,15 +99,53 @@ func snapshotConfig(cfg config.Config) *config.Config { } func (r *Router) SetConfig(cfg config.Config) { + _ = r.setConfig(cfg) +} + +func (r *Router) SetConfigAndWait(ctx context.Context, cfg config.Config) error { + for _, drained := range r.setConfig(cfg) { + select { + case <-drained: + case <-ctx.Done(): + return ctx.Err() + } + } + return nil +} + +func (r *Router) setConfig(cfg config.Config) []<-chan struct{} { config.ApplyDefaults(&cfg) - r.cfg.Store(snapshotConfig(cfg)) + snapshot := snapshotConfig(cfg) r.mu.Lock() + if r.stopping { + r.mu.Unlock() + return nil + } cached := r.runnerCache r.runnerCache = map[string]runner.Runner{} r.activeFallbackDelay = time.Duration(cfg.Active.FallbackDelaySeconds) * time.Second r.activeFallbackLimit = cfg.Active.FallbackBatchLimit + // Publish the new snapshot while the cache rotation lock is still held. + // A handler that observes this generation will then wait for the empty new + // cache instead of inserting its runner into the cache being retired. + r.cfg.Store(snapshot) + retire := len(cached) > 0 + if retire { + // Register under r.mu. Stop takes the same lock before it starts Wait, + // then marks the router stopping so no later Add can race that Wait. + r.retiredWG.Add(1) + } + drained := make([]<-chan struct{}, 0, len(r.configUses)) + for generation, use := range r.configUses { + if generation != snapshot && use.count > 0 { + drained = append(drained, use.done) + } + } r.mu.Unlock() - r.retireRunners(cached) + if retire { + r.retireRunners(cached) + } + return drained } // Stop shuts the router down: it signals scheduled fallback goroutines to @@ -107,16 +153,26 @@ func (r *Router) SetConfig(cfg config.Config) { // background work has drained. After Stop returns with a nil error it is safe // to close the underlying store. func (r *Router) Stop(ctx context.Context) error { - r.stopOnce.Do(func() { close(r.stopped) }) r.mu.Lock() + r.stopping = true + r.stopOnce.Do(func() { close(r.stopped) }) cached := r.runnerCache r.runnerCache = map[string]runner.Runner{} + configDrains := make([]<-chan struct{}, 0, len(r.configUses)) + for _, use := range r.configUses { + if use.count > 0 { + configDrains = append(configDrains, use.done) + } + } r.mu.Unlock() err := stopRunners(ctx, cached) drained := make(chan struct{}) go func() { r.fallbackWG.Wait() r.retiredWG.Wait() + for _, configDrain := range configDrains { + <-configDrain + } close(drained) }() select { @@ -129,11 +185,36 @@ func (r *Router) Stop(ctx context.Context) error { return err } -func (r *Router) retireRunners(cached map[string]runner.Runner) { - if len(cached) == 0 { - return +func (r *Router) acquireConfigSnapshot() (*config.Config, func()) { + r.mu.Lock() + if r.stopping { + r.mu.Unlock() + return nil, func() {} } - r.retiredWG.Add(1) + cfg := r.cfg.Load() + if cfg == nil { + r.mu.Unlock() + return nil, func() {} + } + use := r.configUses[cfg] + if use == nil { + use = &routerConfigUse{done: make(chan struct{})} + r.configUses[cfg] = use + } + use.count++ + r.mu.Unlock() + return cfg, func() { + r.mu.Lock() + use.count-- + if use.count == 0 { + delete(r.configUses, cfg) + close(use.done) + } + r.mu.Unlock() + } +} + +func (r *Router) retireRunners(cached map[string]runner.Runner) { go func() { defer r.retiredWG.Done() if err := stopRunners(context.Background(), cached); err != nil { @@ -146,9 +227,10 @@ func (r *Router) Handle(ctx context.Context, msg types.IncomingMessage) ProcessR lock := r.lockForGroup(msg.ChatID) lock.Lock() defer lock.Unlock() - cfg := r.cfg.Load() + cfg, releaseConfig := r.acquireConfigSnapshot() + defer releaseConfig() if cfg == nil { - return ProcessResult{Ignored: true, Reason: "router config is not initialized"} + return ProcessResult{Ignored: true, Reason: "router config is not initialized or is stopping"} } result, err := r.process(ctx, cfg, msg) if err != nil { @@ -254,7 +336,7 @@ func (r *Router) process(ctx context.Context, cfg *config.Config, msg types.Inco if _, connected, err := r.store.ActiveWatcherFresh(ctx, cfg.App.Profile, sessionID, activeWatcherStaleAfter); err != nil { return ProcessResult{}, err } else if !connected && activeSessionFallbackAllowed(cfg, group) { - if r.activeFallbackDelayValue() <= 0 { + if r.activeFallbackDelayValue(cfg) <= 0 { return r.processActiveSessionFallback(ctx, cfg, msg, group, sessionID) } scheduled := r.scheduleActiveSessionFallback(cfg, msg, group, sessionID) @@ -423,6 +505,9 @@ func (r *Router) scheduleActiveSessionFallback(cfg *config.Config, msg types.Inc key := activeFallbackScheduleKey(cfg.App.Profile, msg.ChatID, sessionID) r.mu.Lock() delay := r.activeFallbackDelay + if r.cfg.Load() != cfg { + delay = time.Duration(cfg.Active.FallbackDelaySeconds) * time.Second + } if delay < 0 { delay = 0 } @@ -481,7 +566,7 @@ func (r *Router) scheduleActiveSessionFallback(cfg *config.Config, msg types.Inc }() // Load a fresh snapshot per firing: config may have been // reloaded while this goroutine waited on its timer. - fireCfg := r.cfg.Load() + fireCfg, releaseConfig := r.acquireConfigSnapshot() if fireCfg == nil { cancel() lock.Unlock() @@ -492,12 +577,14 @@ func (r *Router) scheduleActiveSessionFallback(cfg *config.Config, msg types.Inc result := ProcessResult{Ignored: true, Reason: "active inbox fallback canceled after config reload"} cancel() lock.Unlock() + releaseConfig() r.auditRoute(context.Background(), fireCfg, msg, group, result, map[string]any{"async_fallback": true}) return } result, err := r.processActiveSessionFallback(ctx, fireCfg, msg, fireGroup, sessionID) cancel() lock.Unlock() + releaseConfig() if err != nil { result = ProcessResult{Ignored: true, Reason: err.Error()} } @@ -533,7 +620,7 @@ func (r *Router) processActiveSessionFallback(ctx context.Context, cfg *config.C } else if connected { return ProcessResult{Reason: "active inbox fallback skipped because watcher connected"}, nil } - claimed, err := r.store.ClaimActiveInboxBatchForSession(ctx, cfg.App.Profile, msg.ChatID, sessionID, r.activeFallbackLimitValue()) + claimed, err := r.store.ClaimActiveInboxBatchForSession(ctx, cfg.App.Profile, msg.ChatID, sessionID, r.activeFallbackLimitValue(cfg)) if err != nil { return ProcessResult{}, err } @@ -560,15 +647,21 @@ func (r *Router) processActiveSessionFallback(ctx context.Context, cfg *config.C return result, nil } -func (r *Router) activeFallbackDelayValue() time.Duration { +func (r *Router) activeFallbackDelayValue(cfg *config.Config) time.Duration { r.mu.Lock() defer r.mu.Unlock() + if r.cfg.Load() != cfg { + return time.Duration(cfg.Active.FallbackDelaySeconds) * time.Second + } return r.activeFallbackDelay } -func (r *Router) activeFallbackLimitValue() int { +func (r *Router) activeFallbackLimitValue(cfg *config.Config) int { r.mu.Lock() defer r.mu.Unlock() + if r.cfg.Load() != cfg { + return cfg.Active.FallbackBatchLimit + } return r.activeFallbackLimit } @@ -722,17 +815,84 @@ func (r *Router) runnerFor(cfg *config.Config, chatID, runnerID string, runnerCf key := strings.Join([]string{cfg.App.Profile, chatID, runnerID, fmt.Sprintf("%p", cfg)}, "\x00") r.mu.Lock() defer r.mu.Unlock() + if r.stopping || r.cfg.Load() != cfg { + // This Handle call started on the previous config generation and reached + // runner creation after SetConfig rotated the persistent cache. Let its + // request finish on a private runner, then stop that process instead of + // leaking the old runner into the new generation's cache. + return &singleUseRunner{inner: runner.NewProcessRunner(runnerCfg, cfg.RateLimits.MaxRunnerSeconds)} + } if r.runnerCache == nil { r.runnerCache = map[string]runner.Runner{} } if cached, ok := r.runnerCache[key]; ok { return cached } - cached := runner.NewProcessRunner(runnerCfg, cfg.RateLimits.MaxRunnerSeconds) + cached := newManagedPersistentRunner(runnerCfg, cfg.RateLimits.MaxRunnerSeconds) r.runnerCache[key] = cached return cached } +type managedPersistentRunner struct { + mu sync.Mutex + inner runner.Runner + newRunner func() runner.Runner + retired bool +} + +func newManagedPersistentRunner(runnerCfg config.RunnerConfig, fallbackTimeout int) *managedPersistentRunner { + factory := func() runner.Runner { + return runner.NewProcessRunner(runnerCfg, fallbackTimeout) + } + return &managedPersistentRunner{inner: factory(), newRunner: factory} +} + +func (r *managedPersistentRunner) Invoke(ctx context.Context, req runner.Request) (runner.Result, error) { + r.mu.Lock() + if r.retired { + fresh := r.newRunner() + r.mu.Unlock() + return (&singleUseRunner{inner: fresh}).Invoke(ctx, req) + } + defer r.mu.Unlock() + return r.inner.Invoke(ctx, req) +} + +func (r *managedPersistentRunner) Health(ctx context.Context) error { + r.mu.Lock() + defer r.mu.Unlock() + return r.inner.Health(ctx) +} + +func (r *managedPersistentRunner) Stop(ctx context.Context) error { + r.mu.Lock() + defer r.mu.Unlock() + r.retired = true + return r.inner.Stop(ctx) +} + +type singleUseRunner struct { + inner runner.Runner +} + +func (r *singleUseRunner) Invoke(ctx context.Context, req runner.Request) (runner.Result, error) { + result, err := r.inner.Invoke(ctx, req) + stopCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + if stopErr := r.inner.Stop(stopCtx); stopErr != nil { + fmt.Fprintf(os.Stderr, "stopping single-use runner after config reload: %v\n", stopErr) + } + return result, err +} + +func (r *singleUseRunner) Health(ctx context.Context) error { + return r.inner.Health(ctx) +} + +func (r *singleUseRunner) Stop(ctx context.Context) error { + return r.inner.Stop(ctx) +} + func stopRunners(ctx context.Context, runners map[string]runner.Runner) error { var firstErr error for _, item := range runners { diff --git a/internal/router/router_test.go b/internal/router/router_test.go index cc5a6ad..c4fe272 100644 --- a/internal/router/router_test.go +++ b/internal/router/router_test.go @@ -1801,10 +1801,10 @@ func TestRouterSetConfigUpdatesActiveFallbackSettings(t *testing.T) { updated.Active.FallbackBatchLimit = 3 r.SetConfig(updated) - if delay := r.activeFallbackDelayValue(); delay != 7*time.Second { + if delay := r.activeFallbackDelayValue(r.cfg.Load()); delay != 7*time.Second { t.Fatalf("active fallback delay = %s, want 7s", delay) } - if limit := r.activeFallbackLimitValue(); limit != 3 { + if limit := r.activeFallbackLimitValue(r.cfg.Load()); limit != 3 { t.Fatalf("active fallback limit = %d, want 3", limit) } } @@ -1815,6 +1815,36 @@ type blockingStopRunner struct { stopOnce sync.Once } +type countingRunner struct { + mu sync.Mutex + invokes int + stops int +} + +func (r *countingRunner) Invoke(context.Context, runner.Request) (runner.Result, error) { + r.mu.Lock() + r.invokes++ + r.mu.Unlock() + return runner.Result{}, nil +} + +func (r *countingRunner) Health(context.Context) error { + return nil +} + +func (r *countingRunner) Stop(context.Context) error { + r.mu.Lock() + r.stops++ + r.mu.Unlock() + return nil +} + +func (r *countingRunner) counts() (int, int) { + r.mu.Lock() + defer r.mu.Unlock() + return r.invokes, r.stops +} + func (r *blockingStopRunner) Invoke(context.Context, runner.Request) (runner.Result, error) { return runner.Result{}, nil } @@ -1868,6 +1898,130 @@ func TestRouterSetConfigRetiresCachedRunnersWithoutBlocking(t *testing.T) { } } +func TestRouterSetConfigIsIgnoredAfterStop(t *testing.T) { + t.Parallel() + cfg := config.Default() + r := New(cfg, nil, nil) + before := r.cfg.Load() + if err := r.Stop(t.Context()); err != nil { + t.Fatal(err) + } + updated := cfg + updated.Active.AckMode = "verbose" + r.SetConfig(updated) + if after := r.cfg.Load(); after != before { + t.Fatal("SetConfig published a new generation after Router.Stop") + } +} + +func TestRouterSetConfigAndWaitDrainsPreviousGeneration(t *testing.T) { + t.Parallel() + cfg := config.Default() + r := New(cfg, nil, nil) + previous, releasePrevious := r.acquireConfigSnapshot() + if previous == nil { + t.Fatal("failed to acquire previous config generation") + } + updated := cfg + updated.Active.AckMode = "verbose" + completed := make(chan error, 1) + go func() { + completed <- r.SetConfigAndWait(t.Context(), updated) + }() + select { + case err := <-completed: + t.Fatalf("SetConfigAndWait returned before previous generation drained: %v", err) + case <-time.After(100 * time.Millisecond): + } + releasePrevious() + select { + case err := <-completed: + if err != nil { + t.Fatal(err) + } + case <-time.After(2 * time.Second): + t.Fatal("SetConfigAndWait did not finish after previous generation drained") + } + if r.cfg.Load() == previous { + t.Fatal("new config generation was not published") + } + if err := r.Stop(t.Context()); err != nil { + t.Fatal(err) + } +} + +func TestRouterRunnerForDoesNotCachePreviousConfigGeneration(t *testing.T) { + t.Parallel() + cfg := config.Default() + cfg.App.Profile = "test" + r := New(cfg, nil, nil) + oldSnapshot := r.cfg.Load() + updated := cfg + updated.Active.AckMode = "verbose" + r.SetConfig(updated) + runnerCfg := config.RunnerConfig{ + Mode: "process-jsonl", + Command: os.Args[0], + } + + stale := r.runnerFor(oldSnapshot, "chat@g.us", "persistent", runnerCfg) + if _, ok := stale.(*singleUseRunner); !ok { + t.Fatalf("stale generation runner type = %T, want *singleUseRunner", stale) + } + r.mu.Lock() + cachedAfterStale := len(r.runnerCache) + r.mu.Unlock() + if cachedAfterStale != 0 { + t.Fatalf("stale generation populated current runner cache: %d entries", cachedAfterStale) + } + if err := stale.Stop(t.Context()); err != nil { + t.Fatal(err) + } + + currentSnapshot := r.cfg.Load() + current := r.runnerFor(currentSnapshot, "chat@g.us", "persistent", runnerCfg) + if _, ok := current.(*singleUseRunner); ok { + t.Fatalf("current generation runner type = %T, want cached process runner", current) + } + r.mu.Lock() + cachedAfterCurrent := len(r.runnerCache) + r.mu.Unlock() + if cachedAfterCurrent != 1 { + t.Fatalf("current generation runner cache entries = %d, want 1", cachedAfterCurrent) + } + stopCtx, cancel := context.WithTimeout(t.Context(), 2*time.Second) + defer cancel() + if err := r.Stop(stopCtx); err != nil { + t.Fatal(err) + } +} + +func TestManagedPersistentRunnerCleansLateInvokeAfterRetirement(t *testing.T) { + t.Parallel() + primary := &countingRunner{} + replacement := &countingRunner{} + managed := &managedPersistentRunner{ + inner: primary, + newRunner: func() runner.Runner { return replacement }, + } + // This models runnerFor returning the managed cache entry, followed by a + // config reload retiring it before the caller reaches Invoke. + if err := managed.Stop(t.Context()); err != nil { + t.Fatal(err) + } + if _, err := managed.Invoke(t.Context(), runner.Request{RequestID: "late"}); err != nil { + t.Fatal(err) + } + primaryInvokes, primaryStops := primary.counts() + replacementInvokes, replacementStops := replacement.counts() + if primaryInvokes != 0 || primaryStops != 1 { + t.Fatalf("primary counts invokes=%d stops=%d", primaryInvokes, primaryStops) + } + if replacementInvokes != 1 || replacementStops != 1 { + t.Fatalf("replacement counts invokes=%d stops=%d", replacementInvokes, replacementStops) + } +} + func waitForRouterCondition(t *testing.T, timeout time.Duration, ok func() bool) { t.Helper() deadline := time.Now().Add(timeout) From f890be8933a2c37807462148ccc0ea5d55a28886 Mon Sep 17 00:00:00 2001 From: Dmytro Nikolayev Date: Sat, 11 Jul 2026 17:52:22 +0200 Subject: [PATCH 2/2] fix(config): guard first-run creation --- internal/app/setup.go | 29 ++++++++++--------- internal/config/config.go | 52 +++++++++++++++++++++++++++++----- internal/config/config_test.go | 50 ++++++++++++++++++++++++++++++++ 3 files changed, 110 insertions(+), 21 deletions(-) diff --git a/internal/app/setup.go b/internal/app/setup.go index 3001d1e..ccecaa6 100644 --- a/internal/app/setup.go +++ b/internal/app/setup.go @@ -138,23 +138,24 @@ func (s *cliState) runSetupWizard(cmd *cobra.Command, opts setupWizardOptions) e if err != nil { return err } - if profile := strings.TrimSpace(opts.Profile); profile != "" { - cfg.App.Profile = profile + profileOverride := strings.TrimSpace(opts.Profile) + if profileOverride != "" { + cfg.App.Profile = profileOverride } - if err := config.EnsureProfileDirs(cfg.App.Profile); err != nil { + if err := config.SaveIfMissing(path, cfg); err != nil { return err } - if _, statErr := os.Stat(path); errors.Is(statErr, os.ErrNotExist) { - if err := config.Save(path, cfg); err != nil { - return err - } - // Load the just-created file so the final save is guarded by its exact - // revision. Setup can wait a long time for login or group creation; a - // zero-revision config would otherwise overwrite concurrent updates. - cfg, err = config.Load(path) - if err != nil { - return err - } + // Always continue from the exact on-disk revision. Another first-run + // process may have won creation while this wizard was starting. + cfg, err = config.Load(path) + if err != nil { + return err + } + if profileOverride != "" { + cfg.App.Profile = profileOverride + } + if err := config.EnsureProfileDirs(cfg.App.Profile); err != nil { + return err } workdir := strings.TrimSpace(opts.Workdir) diff --git a/internal/config/config.go b/internal/config/config.go index 7e64ceb..4c5d003 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -35,6 +35,7 @@ type Config struct { Groups []GroupConfig `toml:"groups"` sourcePath string sourceRevision [sha256.Size]byte + sourceMissing bool } var ErrConfigChanged = errors.New("config changed since it was loaded") @@ -255,6 +256,8 @@ func LoadOrDefault(path string) (Config, string, error) { } if errors.Is(err, os.ErrNotExist) { cfg := Default() + cfg.sourcePath = filepath.Clean(path) + cfg.sourceMissing = true return cfg, path, nil } return Config{}, path, err @@ -278,15 +281,50 @@ func Save(path string, cfg Config) error { return saveConfigFile(path, data, cfg) } +// SaveIfMissing atomically creates a config for first-run workflows without +// replacing a file another coderoam process created first. +func SaveIfMissing(path string, cfg Config) error { + if path == "" { + path = DefaultConfigPath() + } + ApplyDefaults(&cfg) + if err := ValidateActiveSessionBindings(cfg); err != nil { + return err + } + if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { + return err + } + data, err := toml.Marshal(cfg) + if err != nil { + return err + } + return withConfigMutationLock(path, func() error { + if _, err := os.Stat(path); err == nil { + return nil + } else if !errors.Is(err, os.ErrNotExist) { + return err + } + return atomicWriteConfig(path, data) + }) +} + func saveConfigFile(path string, data []byte, cfg Config) error { return withConfigMutationLock(path, func() error { - if cfg.sourcePath == filepath.Clean(path) && cfg.sourceRevision != ([sha256.Size]byte{}) { - current, err := os.ReadFile(path) - if err != nil { - return err - } - if sha256.Sum256(current) != cfg.sourceRevision { - return fmt.Errorf("%w at %s; reload and retry", ErrConfigChanged, path) + if cfg.sourcePath == filepath.Clean(path) { + if cfg.sourceMissing { + if _, err := os.Stat(path); err == nil { + return fmt.Errorf("%w at %s; reload and retry", ErrConfigChanged, path) + } else if !errors.Is(err, os.ErrNotExist) { + return err + } + } else if cfg.sourceRevision != ([sha256.Size]byte{}) { + current, err := os.ReadFile(path) + if err != nil { + return err + } + if sha256.Sum256(current) != cfg.sourceRevision { + return fmt.Errorf("%w at %s; reload and retry", ErrConfigChanged, path) + } } } return atomicWriteConfig(path, data) diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 8235561..864a416 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -238,6 +238,56 @@ func TestSaveRejectsStaleLoadedConfig(t *testing.T) { } } +func TestSaveRejectsConfigCreatedAfterLoadOrDefault(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "config.toml") + staleMissing, resolvedPath, err := LoadOrDefault(path) + if err != nil { + t.Fatal(err) + } + if resolvedPath != path { + t.Fatalf("resolved path = %q, want %q", resolvedPath, path) + } + concurrent := Default() + concurrent.Active.AckMode = "verbose" + if err := Save(path, concurrent); err != nil { + t.Fatal(err) + } + staleMissing.Active.AckMode = "off" + if err := Save(path, staleMissing); !errors.Is(err, ErrConfigChanged) { + t.Fatalf("missing-snapshot Save error = %v, want ErrConfigChanged", err) + } + loaded, err := Load(path) + if err != nil { + t.Fatal(err) + } + if loaded.Active.AckMode != "verbose" { + t.Fatalf("missing snapshot replaced concurrent config: ack_mode=%q", loaded.Active.AckMode) + } +} + +func TestSaveIfMissingPreservesExistingConfig(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "config.toml") + existing := Default() + existing.Active.AckMode = "verbose" + if err := Save(path, existing); err != nil { + t.Fatal(err) + } + candidate := Default() + candidate.Active.AckMode = "off" + if err := SaveIfMissing(path, candidate); err != nil { + t.Fatal(err) + } + loaded, err := Load(path) + if err != nil { + t.Fatal(err) + } + if loaded.Active.AckMode != "verbose" { + t.Fatalf("SaveIfMissing replaced existing config: ack_mode=%q", loaded.Active.AckMode) + } +} + func TestSaveNeverExposesPartialConfig(t *testing.T) { t.Parallel() path := filepath.Join(t.TempDir(), "config.toml")