From 1019aa0321490149776e12e69542a58e38216f48 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Thu, 11 Jun 2026 00:09:59 +0100 Subject: [PATCH] fix: fail fast messages sync behind live tail --- CHANGELOG.md | 1 + README.md | 2 +- docs/commands/messages.md | 3 +- internal/cli/admin_commands.go | 6 + internal/cli/cli.go | 73 ++-- internal/cli/cli_test.go | 482 ++++++++++++++++++++++++- internal/cli/output.go | 2 +- internal/cli/query_sync.go | 10 +- internal/cli/query_sync_test.go | 6 + internal/cli/sync_lock.go | 292 +++++++++++++-- internal/cli/sync_lock_other.go | 8 + internal/cli/sync_lock_pid_other.go | 7 + internal/cli/sync_lock_pid_unix.go | 13 + internal/cli/sync_lock_pid_windows.go | 17 + internal/cli/sync_lock_unix.go | 24 +- internal/cli/sync_lock_windows.go | 32 +- internal/cli/sync_lock_windows_test.go | 12 + internal/discord/client.go | 9 + internal/discord/client_test.go | 7 + internal/syncer/syncer.go | 9 + internal/syncer/syncer_tail_test.go | 75 ++++ internal/syncer/tail.go | 18 +- 22 files changed, 1036 insertions(+), 72 deletions(-) create mode 100644 internal/cli/sync_lock_pid_other.go create mode 100644 internal/cli/sync_lock_pid_unix.go create mode 100644 internal/cli/sync_lock_pid_windows.go create mode 100644 internal/cli/sync_lock_windows_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 75ad4d27..8deca8f1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +24,7 @@ ### Fixes - Kept resumed `sync --full` backfills from moving channel latest-message checkpoints backward, avoiding duplicate head recrawls on large interrupted channels. Thanks @hannesrudolph. +- Made `messages --sync` fail fast with an omit-`--sync` hint when a live `tail` process owns the sync lock, while plain `messages` reads continue without waiting. Thanks @jeanmonet. ## 0.9.1 - 2026-05-18 diff --git a/README.md b/README.md index 9fd13fc7..ec190bfa 100644 --- a/README.md +++ b/README.md @@ -384,7 +384,7 @@ Notes: - `--days` is shorthand for "since now minus N days" - `--last` returns the newest `N` matching messages, then prints them oldest-to-newest - `--all` removes the safety limit; default is `200` -- `--sync` runs a blocking pre-query sync for the matching channel or guild scope before reading the local DB +- `--sync` runs a blocking pre-query sync for the matching channel or guild scope before reading the local DB; omit it while `tail` is already maintaining live freshness - rows with no displayable/searchable content are skipped by default; `--include-empty` opts back in - at least one filter is required - `--dm` is shorthand for `--guild @me`, so DM searches and message slices do not need raw SQL diff --git a/docs/commands/messages.md b/docs/commands/messages.md index 7d2fb61a..b447a8f3 100644 --- a/docs/commands/messages.md +++ b/docs/commands/messages.md @@ -26,12 +26,13 @@ discrawl --json messages --channel maintainers --days 3 - `--last ` - return the newest `N` matching messages, then print oldest-to-newest - `--limit ` - safety limit (default 200; `--all` removes it) - `--all` - removes the safety limit -- `--sync` - blocking pre-query sync for the matching channel or guild scope +- `--sync` - blocking pre-query sync for the matching channel or guild scope; omit while `tail` is already maintaining live freshness - `--include-empty` - include rows with no displayable/searchable content ## Notes - at least one filter is required +- if `tail` is already running, plain `messages` reads the local archive without waiting; `messages --sync` fails fast instead of waiting behind the tail lock - `--dm` skips Git snapshot auto-update because DMs are never imported from the shared mirror - use either `--last` for the newest matching rows or `--all` for an uncapped oldest-to-newest slice diff --git a/internal/cli/admin_commands.go b/internal/cli/admin_commands.go index eae09bdf..6384b4dc 100644 --- a/internal/cli/admin_commands.go +++ b/internal/cli/admin_commands.go @@ -332,6 +332,12 @@ func (r *runtime) runTail(args []string) error { } ctx, stop := signal.NotifyContext(r.ctx, os.Interrupt, syscall.SIGTERM) defer stop() + if configurable, ok := r.syncer.(tailReadyConfigurer); ok { + configurable.SetTailReadyCallback(func(context.Context) error { + return r.activateTailSyncLock() + }) + defer configurable.SetTailReadyCallback(nil) + } return r.syncer.RunTail(ctx, r.resolveSyncGuilds(*guildFlag, *guildsFlag), *repairEvery) } diff --git a/internal/cli/cli.go b/internal/cli/cli.go index e3ee9d06..532dfe5f 100644 --- a/internal/cli/cli.go +++ b/internal/cli/cli.go @@ -145,25 +145,28 @@ func parseKongArgs(target any, args []string, name string, stdout, stderr io.Wri } type runtime struct { - ctx context.Context - configPath string - cfg config.Config - stdout io.Writer - stderr io.Writer - json bool - plain bool - logger *slog.Logger - store *store.Store - client discordClient - syncer syncService - dbLockHeld bool - lockStarted time.Time - openStore func(context.Context, string) (*store.Store, error) - newDiscord func(config.Config) (discordClient, error) - newRemote func(config.Config) (remoteArchiveClient, error) - newSyncer func(syncer.Client, *store.Store, *slog.Logger) syncService - newEmbed func(config.EmbeddingsConfig) (embed.Provider, error) - now func() time.Time + ctx context.Context + configPath string + cfg config.Config + stdout io.Writer + stderr io.Writer + json bool + plain bool + logger *slog.Logger + store *store.Store + client discordClient + syncer syncService + dbLockHeld bool + lockStarted time.Time + lockOperation string + lockToken string + lockTokenFree func() error + openStore func(context.Context, string) (*store.Store, error) + newDiscord func(config.Config) (discordClient, error) + newRemote func(config.Config) (remoteArchiveClient, error) + newSyncer func(syncer.Client, *store.Store, *slog.Logger) syncService + newEmbed func(config.EmbeddingsConfig) (embed.Provider, error) + now func() time.Time } func crawlkitEmbeddingConfig(cfg config.EmbeddingsConfig) embed.Config { @@ -190,6 +193,10 @@ type syncService interface { RunTail(context.Context, []string, time.Duration) error } +type tailReadyConfigurer interface { + SetTailReadyCallback(func(context.Context) error) +} + type attachmentTextConfigurer interface { SetAttachmentTextEnabled(bool) } @@ -209,7 +216,7 @@ func (r *runtime) dispatch(rest []string) error { } return r.withLocalStoreUpdateLocked(updateMode, true, func() error { return r.runSync(rest[1:]) }) case "tail": - return r.withServicesLocked(true, func() error { return r.runTail(rest[1:]) }) + return r.withServicesLockedOperation(true, "tail-starting", func() error { return r.runTail(rest[1:]) }) case "wiretap": return r.withLocalStoreLocked(false, func() error { return r.runWiretap(rest[1:]) }) case "tap", "cache-import": @@ -235,10 +242,10 @@ func (r *runtime) dispatch(rest []string) error { if r.configuredForCloudReadOnly() { return r.withConfig(func() error { return r.runMessages(rest[1:]) }) } - if hasBoolFlag(rest[1:], "--sync") && !hasBoolFlag(rest[1:], "--dm") { - return r.withServicesAutoLocked(true, true, true, func() error { return r.runMessages(rest[1:]) }) + if boolFlagEnabled(rest[1:], "--sync") && !boolFlagEnabled(rest[1:], "--dm") { + return r.withMessagesSyncServices(func() error { return r.runMessages(rest[1:]) }) } - autoShareUpdate := !hasBoolFlag(rest[1:], "--dm") + autoShareUpdate := !boolFlagEnabled(rest[1:], "--dm") return r.withLocalStoreRead(autoShareUpdate, func() error { return r.runMessages(rest[1:]) }) case "digest": return r.withLocalStoreRead(true, func() error { return r.runDigest(rest[1:]) }) @@ -323,8 +330,8 @@ func (r *runtime) withServices(withDiscord bool, fn func() error) error { return r.withServicesAuto(withDiscord, !withDiscord, fn) } -func (r *runtime) withServicesLocked(withDiscord bool, fn func() error) error { - return r.withServicesAutoLocked(withDiscord, !withDiscord, true, fn) +func (r *runtime) withServicesLockedOperation(withDiscord bool, operation string, fn func() error) error { + return r.withServicesUpdateLockedOperation(withDiscord, boolShareUpdateMode(!withDiscord), true, operation, fn) } func (r *runtime) withLocalStoreLocked(autoShareUpdate bool, fn func() error) error { @@ -541,10 +548,14 @@ func (r *runtime) withServicesAuto(withDiscord, autoShareUpdate bool, fn func() } func (r *runtime) withServicesAutoLocked(withDiscord, autoShareUpdate, lockDB bool, fn func() error) error { - return r.withServicesUpdateLocked(withDiscord, boolShareUpdateMode(autoShareUpdate), lockDB, fn) + return r.withServicesUpdateLockedOperation(withDiscord, boolShareUpdateMode(autoShareUpdate), lockDB, "writer", fn) +} + +func (r *runtime) withMessagesSyncServices(fn func() error) error { + return r.withServicesUpdateLockedOperation(true, shareUpdateConfigured, true, "messages-sync", fn) } -func (r *runtime) withServicesUpdateLocked(withDiscord bool, updateMode shareUpdateMode, lockDB bool, fn func() error) error { +func (r *runtime) withServicesUpdateLockedOperation(withDiscord bool, updateMode shareUpdateMode, lockDB bool, operation string, fn func() error) error { cfg, err := config.Load(r.configPath) if err != nil { return configErr(err) @@ -558,7 +569,13 @@ func (r *runtime) withServicesUpdateLocked(withDiscord bool, updateMode shareUpd } r.cfg = cfg if lockDB { - return r.withSyncLock(func() error { + lockFn := r.withSyncLockOperation + if operation == "messages-sync" { + lockFn = func(_ string, fn func() error) error { + return r.withMessagesSyncLock(fn) + } + } + return lockFn(operation, func() error { return r.openServices(dbPath, withDiscord, updateMode, fn) }) } diff --git a/internal/cli/cli_test.go b/internal/cli/cli_test.go index 5b874b0f..7fd6608c 100644 --- a/internal/cli/cli_test.go +++ b/internal/cli/cli_test.go @@ -8,6 +8,7 @@ import ( "encoding/json" "errors" "flag" + "fmt" "io" "log/slog" "net/http" @@ -2045,6 +2046,394 @@ func TestReadCommandsDoNotWaitForSyncLock(t *testing.T) { } } +func TestMessagesSyncFailsFastWhenTailOwnsSyncLock(t *testing.T) { + if goruntime.GOOS == "windows" { + t.Skip("sync lock timing is flaky on Windows") + } + ctx := context.Background() + dir := t.TempDir() + cfg, cfgPath := writeTestConfig(t, dir) + s := seedCLIStore(t, cfg.DBPath) + require.NoError(t, s.Close()) + t.Setenv(config.DefaultTokenEnv, "env-token") + + lockPath := filepath.Join(dir, ".discrawl-sync.lock") + release, err := acquireSyncLock(ctx, lockPath) + require.NoError(t, err) + defer func() { _ = release() }() + releaseToken := holdSyncLockToken(t, ctx, lockPath, testSyncLockToken()) + defer releaseToken() + writeSyncLockMetadata(t, lockPath, "tail", os.Getpid()) + + rt, fakeSync := messagesSyncTestRuntime(ctx, cfgPath) + err = rt.dispatch([]string{"messages", "--channel", "general", "--last", "1", "--sync"}) + require.Error(t, err) + require.Equal(t, 2, ExitCode(err)) + require.Contains(t, err.Error(), "tail already owns live sync; omit --sync while tail is running") + require.Zero(t, fakeSync.syncCalls) +} + +func TestMessagesSyncFailsFastDuringTailLockMetadataStartup(t *testing.T) { + if goruntime.GOOS == "windows" { + t.Skip("sync lock timing is flaky on Windows") + } + ctx := context.Background() + dir := t.TempDir() + cfg, cfgPath := writeTestConfig(t, dir) + s := seedCLIStore(t, cfg.DBPath) + require.NoError(t, s.Close()) + t.Setenv(config.DefaultTokenEnv, "env-token") + + lockPath := filepath.Join(dir, ".discrawl-sync.lock") + release, err := acquireSyncLock(ctx, lockPath) + require.NoError(t, err) + defer func() { _ = release() }() + releaseToken := holdSyncLockToken(t, ctx, lockPath, testSyncLockToken()) + defer releaseToken() + require.NoError(t, writeSyncLockMetadataFiles(lockPath, fmt.Appendf(nil, "pid=%d\n", os.Getpid()))) + go func() { + time.Sleep(25 * time.Millisecond) + body := fmt.Sprintf("pid=%d\noperation=tail\ntoken=%s\nstarted_at=2026-03-08T12:00:00Z\nupdated_at=2026-03-08T12:00:00Z\nphase=locked\n", os.Getpid(), testSyncLockToken()) + _ = writeSyncLockMetadataFiles(lockPath, []byte(body)) + }() + + rt, fakeSync := messagesSyncTestRuntime(ctx, cfgPath) + err = rt.dispatch([]string{"messages", "--channel", "general", "--last", "1", "--sync"}) + require.Error(t, err) + require.Equal(t, 2, ExitCode(err)) + require.Contains(t, err.Error(), "tail already owns live sync; omit --sync while tail is running") + require.Zero(t, fakeSync.syncCalls) +} + +func TestMessagesSyncIgnoresStaleTailLockMetadata(t *testing.T) { + if goruntime.GOOS == "windows" { + t.Skip("sync lock timing is flaky on Windows") + } + ctx := context.Background() + dir := t.TempDir() + cfg, cfgPath := writeTestConfig(t, dir) + s := seedCLIStore(t, cfg.DBPath) + require.NoError(t, s.Close()) + t.Setenv(config.DefaultTokenEnv, "env-token") + + writeSyncLockMetadata(t, filepath.Join(dir, ".discrawl-sync.lock"), "tail", os.Getpid()) + + rt, fakeSync := messagesSyncTestRuntime(ctx, cfgPath) + require.NoError(t, rt.dispatch([]string{"messages", "--channel", "general", "--last", "1", "--sync"})) + require.Equal(t, 1, fakeSync.syncCalls) + require.Contains(t, rt.stdout.(*bytes.Buffer).String(), "automatic updates work") +} + +func TestMessagesSyncTreatsStaleTailMetadataHeldByNonTailAsWriter(t *testing.T) { + if goruntime.GOOS == "windows" { + t.Skip("sync lock timing is flaky on Windows") + } + ctx := context.Background() + dir := t.TempDir() + cfg, cfgPath := writeTestConfig(t, dir) + s := seedCLIStore(t, cfg.DBPath) + require.NoError(t, s.Close()) + t.Setenv(config.DefaultTokenEnv, "env-token") + + lockPath := filepath.Join(dir, ".discrawl-sync.lock") + release, err := acquireSyncLock(ctx, lockPath) + require.NoError(t, err) + defer func() { _ = release() }() + writeSyncLockMetadata(t, lockPath, "tail", os.Getpid()) + + waitCtx, cancel := context.WithTimeout(ctx, 25*time.Millisecond) + defer cancel() + rt, fakeSync := messagesSyncTestRuntime(waitCtx, cfgPath) + err = rt.dispatch([]string{"messages", "--channel", "general", "--last", "1", "--sync"}) + require.ErrorIs(t, err, context.DeadlineExceeded) + require.NotContains(t, err.Error(), "tail already owns live sync") + require.Zero(t, fakeSync.syncCalls) +} + +func TestMessagesSyncWaitsDuringTailStartup(t *testing.T) { + if goruntime.GOOS == "windows" { + t.Skip("sync lock timing is flaky on Windows") + } + ctx := context.Background() + dir := t.TempDir() + cfg, cfgPath := writeTestConfig(t, dir) + s := seedCLIStore(t, cfg.DBPath) + require.NoError(t, s.Close()) + t.Setenv(config.DefaultTokenEnv, "env-token") + + lockPath := filepath.Join(dir, ".discrawl-sync.lock") + release, err := acquireSyncLockWithMetadata(ctx, lockPath, syncLockMetadataBody("tail-starting", "locked", time.Now().UTC(), time.Now().UTC(), testSyncLockToken())) + require.NoError(t, err) + defer func() { _ = release() }() + + waitCtx, cancel := context.WithTimeout(ctx, 25*time.Millisecond) + defer cancel() + rt, fakeSync := messagesSyncTestRuntime(waitCtx, cfgPath) + err = rt.dispatch([]string{"messages", "--channel", "general", "--last", "1", "--sync"}) + require.ErrorIs(t, err, context.DeadlineExceeded) + require.NotContains(t, err.Error(), "tail already owns live sync") + require.Zero(t, fakeSync.syncCalls) +} + +func TestMessagesSyncWaitsForNonTailSyncLockOwner(t *testing.T) { + if goruntime.GOOS == "windows" { + t.Skip("sync lock timing is flaky on Windows") + } + ctx := context.Background() + dir := t.TempDir() + cfg, cfgPath := writeTestConfig(t, dir) + s := seedCLIStore(t, cfg.DBPath) + require.NoError(t, s.Close()) + t.Setenv(config.DefaultTokenEnv, "env-token") + + lockPath := filepath.Join(dir, ".discrawl-sync.lock") + release, err := acquireSyncLock(ctx, lockPath) + require.NoError(t, err) + defer func() { _ = release() }() + writeSyncLockMetadata(t, lockPath, "sync", os.Getpid()) + + waitCtx, cancel := context.WithTimeout(ctx, 25*time.Millisecond) + defer cancel() + rt, fakeSync := messagesSyncTestRuntime(waitCtx, cfgPath) + err = rt.dispatch([]string{"messages", "--channel", "general", "--last", "1", "--sync"}) + require.ErrorIs(t, err, context.DeadlineExceeded) + require.Zero(t, fakeSync.syncCalls) +} + +func TestMessagesSyncWaitsForLegacySyncLockMetadata(t *testing.T) { + if goruntime.GOOS == "windows" { + t.Skip("sync lock timing is flaky on Windows") + } + ctx := context.Background() + dir := t.TempDir() + cfg, cfgPath := writeTestConfig(t, dir) + s := seedCLIStore(t, cfg.DBPath) + require.NoError(t, s.Close()) + t.Setenv(config.DefaultTokenEnv, "env-token") + + lockPath := filepath.Join(dir, ".discrawl-sync.lock") + release, err := acquireSyncLock(ctx, lockPath) + require.NoError(t, err) + defer func() { _ = release() }() + require.NoError(t, os.Remove(syncLockMetadataPath(lockPath))) + require.NoError(t, os.WriteFile(lockPath, fmt.Appendf(nil, "pid=%d\n", os.Getpid()), 0o600)) + + waitCtx, cancel := context.WithTimeout(ctx, 25*time.Millisecond) + defer cancel() + rt, fakeSync := messagesSyncTestRuntime(waitCtx, cfgPath) + err = rt.dispatch([]string{"messages", "--channel", "general", "--last", "1", "--sync"}) + require.ErrorIs(t, err, context.DeadlineExceeded) + require.Zero(t, fakeSync.syncCalls) +} + +func TestMessagesSyncWaitsForMalformedSyncLockMetadata(t *testing.T) { + if goruntime.GOOS == "windows" { + t.Skip("sync lock timing is flaky on Windows") + } + ctx := context.Background() + dir := t.TempDir() + cfg, cfgPath := writeTestConfig(t, dir) + s := seedCLIStore(t, cfg.DBPath) + require.NoError(t, s.Close()) + t.Setenv(config.DefaultTokenEnv, "env-token") + + lockPath := filepath.Join(dir, ".discrawl-sync.lock") + release, err := acquireSyncLock(ctx, lockPath) + require.NoError(t, err) + defer func() { _ = release() }() + require.NoError(t, writeSyncLockMetadataFiles(lockPath, []byte("pid\noperation=tail\n"))) + + waitCtx, cancel := context.WithTimeout(ctx, 25*time.Millisecond) + defer cancel() + rt, fakeSync := messagesSyncTestRuntime(waitCtx, cfgPath) + err = rt.dispatch([]string{"messages", "--channel", "general", "--last", "1", "--sync"}) + require.ErrorIs(t, err, context.DeadlineExceeded) + require.Zero(t, fakeSync.syncCalls) +} + +func TestMessagesSyncPreservesCancellationWhileWaitingForSyncLock(t *testing.T) { + if goruntime.GOOS == "windows" { + t.Skip("sync lock timing is flaky on Windows") + } + ctx := context.Background() + dir := t.TempDir() + cfg, cfgPath := writeTestConfig(t, dir) + s := seedCLIStore(t, cfg.DBPath) + require.NoError(t, s.Close()) + t.Setenv(config.DefaultTokenEnv, "env-token") + + lockPath := filepath.Join(dir, ".discrawl-sync.lock") + release, err := acquireSyncLock(ctx, lockPath) + require.NoError(t, err) + defer func() { _ = release() }() + writeSyncLockMetadata(t, lockPath, "sync", os.Getpid()) + + canceledCtx, cancel := context.WithCancel(ctx) + cancel() + rt, fakeSync := messagesSyncTestRuntime(canceledCtx, cfgPath) + err = rt.dispatch([]string{"messages", "--channel", "general", "--last", "1", "--sync"}) + require.ErrorIs(t, err, context.Canceled) + require.Zero(t, fakeSync.syncCalls) +} + +func TestPlainMessagesStillReadsWhileTailOwnsSyncLock(t *testing.T) { + if goruntime.GOOS == "windows" { + t.Skip("sync lock timing is flaky on Windows") + } + ctx := context.Background() + dir := t.TempDir() + cfg, cfgPath := writeTestConfig(t, dir) + s := seedCLIStore(t, cfg.DBPath) + require.NoError(t, s.Close()) + + lockPath := filepath.Join(dir, ".discrawl-sync.lock") + release, err := acquireSyncLock(ctx, lockPath) + require.NoError(t, err) + defer func() { _ = release() }() + releaseToken := holdSyncLockToken(t, ctx, lockPath, testSyncLockToken()) + defer releaseToken() + writeSyncLockMetadata(t, lockPath, "tail", os.Getpid()) + + runCtx, cancel := context.WithTimeout(ctx, 2*time.Second) + defer cancel() + var out bytes.Buffer + err = Run(runCtx, []string{"--config", cfgPath, "messages", "--channel", "general", "--last", "1"}, &out, &bytes.Buffer{}) + require.NoError(t, err) + require.Contains(t, out.String(), "automatic updates work") +} + +func TestMessagesSyncFalseStillReadsWhileTailOwnsSyncLock(t *testing.T) { + if goruntime.GOOS == "windows" { + t.Skip("sync lock timing is flaky on Windows") + } + ctx := context.Background() + dir := t.TempDir() + cfg, cfgPath := writeTestConfig(t, dir) + s := seedCLIStore(t, cfg.DBPath) + require.NoError(t, s.Close()) + + lockPath := filepath.Join(dir, ".discrawl-sync.lock") + release, err := acquireSyncLock(ctx, lockPath) + require.NoError(t, err) + defer func() { _ = release() }() + releaseToken := holdSyncLockToken(t, ctx, lockPath, testSyncLockToken()) + defer releaseToken() + writeSyncLockMetadata(t, lockPath, "tail", os.Getpid()) + + runCtx, cancel := context.WithTimeout(ctx, 2*time.Second) + defer cancel() + var out bytes.Buffer + err = Run(runCtx, []string{"--config", cfgPath, "messages", "--channel", "general", "--last", "1", "--sync=false"}, &out, &bytes.Buffer{}) + require.NoError(t, err) + require.Contains(t, out.String(), "automatic updates work") +} + +func TestMessagesRepeatedSyncFalseStillReadsWhileTailOwnsSyncLock(t *testing.T) { + if goruntime.GOOS == "windows" { + t.Skip("sync lock timing is flaky on Windows") + } + ctx := context.Background() + dir := t.TempDir() + cfg, cfgPath := writeTestConfig(t, dir) + s := seedCLIStore(t, cfg.DBPath) + require.NoError(t, s.Close()) + + lockPath := filepath.Join(dir, ".discrawl-sync.lock") + release, err := acquireSyncLock(ctx, lockPath) + require.NoError(t, err) + defer func() { _ = release() }() + releaseToken := holdSyncLockToken(t, ctx, lockPath, testSyncLockToken()) + defer releaseToken() + writeSyncLockMetadata(t, lockPath, "tail", os.Getpid()) + + runCtx, cancel := context.WithTimeout(ctx, 2*time.Second) + defer cancel() + var out bytes.Buffer + err = Run(runCtx, []string{"--config", cfgPath, "messages", "--channel", "general", "--last", "1", "--sync", "--sync=false"}, &out, &bytes.Buffer{}) + require.NoError(t, err) + require.Contains(t, out.String(), "automatic updates work") +} + +func TestTailOpenFailureDoesNotPublishActiveTailOwner(t *testing.T) { + if goruntime.GOOS == "windows" { + t.Skip("sync lock timing is flaky on Windows") + } + ctx := context.Background() + dir := t.TempDir() + cfg, cfgPath := writeTestConfig(t, dir) + s := seedCLIStore(t, cfg.DBPath) + require.NoError(t, s.Close()) + t.Setenv(config.DefaultTokenEnv, "env-token") + + fakeSync := &fakeSyncService{tailErr: errors.New("gateway open failed")} + rt := tailTestRuntime(ctx, cfgPath, fakeSync) + err := rt.dispatch([]string{"tail"}) + require.ErrorContains(t, err, "gateway open failed") + require.Zero(t, fakeSync.tailReadyCalls) + + lockPath := filepath.Join(dir, ".discrawl-sync.lock") + owner, ok := readSyncLockOwner(lockPath) + require.True(t, ok) + require.Equal(t, "tail-starting", owner.Operation) + require.False(t, rt.activeTailOwnsSyncLock(lockPath)) + require.Empty(t, syncLockOwnerFiles(t, lockPath)) +} + +func TestTailReadyPromotesOnceAndCleansUpOnCancellation(t *testing.T) { + if goruntime.GOOS == "windows" { + t.Skip("sync lock timing is flaky on Windows") + } + ctx := context.Background() + dir := t.TempDir() + cfg, cfgPath := writeTestConfig(t, dir) + s := seedCLIStore(t, cfg.DBPath) + require.NoError(t, s.Close()) + t.Setenv(config.DefaultTokenEnv, "env-token") + + fakeSync := &fakeSyncService{callTailReady: true, tailErr: context.Canceled} + rt := tailTestRuntime(ctx, cfgPath, fakeSync) + err := rt.dispatch([]string{"tail"}) + require.ErrorIs(t, err, context.Canceled) + require.Equal(t, 1, fakeSync.tailReadyCalls) + + lockPath := filepath.Join(dir, ".discrawl-sync.lock") + owner, ok := readSyncLockOwner(lockPath) + require.True(t, ok) + require.Equal(t, "tail", owner.Operation) + require.False(t, rt.activeTailOwnsSyncLock(lockPath)) + require.Empty(t, syncLockOwnerFiles(t, lockPath)) +} + +func TestSyncLockHelperEdges(t *testing.T) { + ctx := context.Background() + dir := t.TempDir() + lockPath := filepath.Join(dir, ".discrawl-sync.lock") + + require.False(t, validSyncLockToken("")) + require.False(t, validSyncLockToken("not-hex-not-hex-not-hex-not-hex")) + require.True(t, validSyncLockToken(testSyncLockToken())) + require.False(t, syncLockTokenHeld(lockPath, testSyncLockToken())) + + require.NoError(t, os.WriteFile(lockPath, []byte("pid=bad\n"), 0o600)) + _, ok := readSyncLockOwner(lockPath) + require.False(t, ok) + + require.NoError(t, writeSyncLockMetadataFiles(lockPath, []byte("pid=123\noperation=legacy\n"))) + owner, ok := readSyncLockOwner(lockPath) + require.True(t, ok) + require.Equal(t, "legacy", owner.Operation) + + require.NoError(t, writeSyncLockMetadataSidecar(lockPath, []byte("pid=123\noperation=sidecar\nphase=current\n"))) + canceledCtx, cancel := context.WithCancel(ctx) + cancel() + err := syncLockErr(canceledCtx, lockPath) + require.ErrorIs(t, err, context.Canceled) + require.Contains(t, err.Error(), "phase=current") + + rt := &runtime{} + require.NoError(t, rt.activateTailSyncLock()) +} + func TestReadCommandsMigrateOlderLocalStore(t *testing.T) { ctx := context.Background() dir := t.TempDir() @@ -2119,6 +2508,78 @@ func seedCLIStore(t *testing.T, path string) *store.Store { return s } +func writeTestConfig(t *testing.T, dir string) (config.Config, string) { + t.Helper() + cfg := config.Default() + cfg.DBPath = filepath.Join(dir, "discrawl.db") + cfg.DefaultGuildID = "g1" + cfgPath := filepath.Join(dir, "config.toml") + require.NoError(t, config.Write(cfgPath, cfg)) + return cfg, cfgPath +} + +func testSyncLockToken() string { + return fmt.Sprintf("%032x", 1) +} + +func writeSyncLockMetadata(t *testing.T, path, operation string, pid int) { + t.Helper() + body := fmt.Sprintf("pid=%d\noperation=%s\ntoken=%s\nstarted_at=2026-03-08T12:00:00Z\nupdated_at=2026-03-08T12:00:00Z\nphase=locked\n", pid, operation, testSyncLockToken()) + require.NoError(t, writeSyncLockMetadataFiles(path, []byte(body))) +} + +func holdSyncLockToken(t *testing.T, ctx context.Context, lockPath, token string) func() { + t.Helper() + now := time.Date(2026, 3, 8, 12, 0, 0, 0, time.UTC) + tokenPath := syncLockTokenPath(lockPath, token) + release, err := acquireSyncLockWithMetadata(ctx, tokenPath, syncLockMetadataBody("tail-token", "locked", now, now, token)) + require.NoError(t, err) + return func() { + _ = release() + _ = os.Remove(tokenPath) + } +} + +func messagesSyncTestRuntime(ctx context.Context, cfgPath string) (*runtime, *fakeSyncService) { + fakeSync := &fakeSyncService{} + out := &bytes.Buffer{} + rt := &runtime{ + ctx: ctx, + configPath: cfgPath, + stdout: out, + stderr: &bytes.Buffer{}, + logger: discardLogger(), + openStore: store.Open, + newDiscord: func(config.Config) (discordClient, error) { return &fakeDiscordClient{}, nil }, + newSyncer: func(syncer.Client, *store.Store, *slog.Logger) syncService { + return fakeSync + }, + } + return rt, fakeSync +} + +func tailTestRuntime(ctx context.Context, cfgPath string, fakeSync *fakeSyncService) *runtime { + return &runtime{ + ctx: ctx, + configPath: cfgPath, + stdout: &bytes.Buffer{}, + stderr: &bytes.Buffer{}, + logger: discardLogger(), + openStore: store.Open, + newDiscord: func(config.Config) (discordClient, error) { return &fakeDiscordClient{}, nil }, + newSyncer: func(syncer.Client, *store.Store, *slog.Logger) syncService { + return fakeSync + }, + } +} + +func syncLockOwnerFiles(t *testing.T, lockPath string) []string { + t.Helper() + files, err := filepath.Glob(lockPath + ".*.owner*") + require.NoError(t, err) + return files +} + func addCLIAttachment(ctx context.Context, s *store.Store, url string) error { now := time.Now().UTC().Format(time.RFC3339Nano) return s.UpsertMessages(ctx, []store.MessageMutation{{ @@ -2612,9 +3073,14 @@ func (f *fakeDiscordClient) Tail(context.Context, discordclient.EventHandler) er type fakeSyncService struct { discovered []*discordgo.UserGuild lastSync syncer.SyncOptions + syncCalls int lastTail []string lastRepair time.Duration attachmentTextEnabled bool + callTailReady bool + tailReadyCalls int + tailReady func(context.Context) error + tailErr error } func (f *fakeSyncService) DiscoverGuilds(context.Context) ([]*discordgo.UserGuild, error) { @@ -2622,16 +3088,30 @@ func (f *fakeSyncService) DiscoverGuilds(context.Context) ([]*discordgo.UserGuil } func (f *fakeSyncService) Sync(_ context.Context, opts syncer.SyncOptions) (syncer.SyncStats, error) { + f.syncCalls++ f.lastSync = opts return syncer.SyncStats{Guilds: len(opts.GuildIDs), Messages: 3}, nil } -func (f *fakeSyncService) RunTail(_ context.Context, guildIDs []string, repairEvery time.Duration) error { +func (f *fakeSyncService) RunTail(ctx context.Context, guildIDs []string, repairEvery time.Duration) error { f.lastTail = guildIDs f.lastRepair = repairEvery + if f.callTailReady && f.tailReady != nil { + f.tailReadyCalls++ + if err := f.tailReady(ctx); err != nil { + return err + } + } + if f.tailErr != nil { + return f.tailErr + } return nil } +func (f *fakeSyncService) SetTailReadyCallback(fn func(context.Context) error) { + f.tailReady = fn +} + func (f *fakeSyncService) SetAttachmentTextEnabled(enabled bool) { f.attachmentTextEnabled = enabled } diff --git a/internal/cli/output.go b/internal/cli/output.go index b22e5b8b..8eaf9196 100644 --- a/internal/cli/output.go +++ b/internal/cli/output.go @@ -201,7 +201,7 @@ Flags: --limit N Maximum rows. Default: 200. --last N Most recent N rows. --all Return all matching rows. - --sync Refresh channel before reading. + --sync Refresh channel before reading; omit while tail is running. --include-empty Include empty/attachment-only messages. --dm Read local desktop DM cache. --guild ID Restrict to one guild id. diff --git a/internal/cli/query_sync.go b/internal/cli/query_sync.go index d6c40b44..507f2be5 100644 --- a/internal/cli/query_sync.go +++ b/internal/cli/query_sync.go @@ -98,18 +98,22 @@ func hasBoolFlag(args []string, name string) bool { } func boolFlagEnabled(args []string, name string) bool { + enabled := false for _, arg := range args { if arg == name { - return true + enabled = true + continue } if raw, ok := strings.CutPrefix(arg, name+"="); ok { switch strings.ToLower(strings.TrimSpace(raw)) { case "1", "t", "true", "y", "yes", "on": - return true + enabled = true + default: + enabled = false } } } - return false + return enabled } func hasHelpArg(args []string) bool { diff --git a/internal/cli/query_sync_test.go b/internal/cli/query_sync_test.go index 85dd5821..6e9c0a9f 100644 --- a/internal/cli/query_sync_test.go +++ b/internal/cli/query_sync_test.go @@ -113,6 +113,12 @@ func TestHasBoolFlag(t *testing.T) { require.True(t, hasBoolFlag([]string{"--sync"}, "--sync")) require.True(t, hasBoolFlag([]string{"--sync=true"}, "--sync")) require.False(t, hasBoolFlag([]string{"--other"}, "--sync")) + require.True(t, boolFlagEnabled([]string{"--sync"}, "--sync")) + require.True(t, boolFlagEnabled([]string{"--sync=true"}, "--sync")) + require.False(t, boolFlagEnabled([]string{"--sync=false"}, "--sync")) + require.False(t, boolFlagEnabled([]string{"--sync", "--sync=false"}, "--sync")) + require.True(t, boolFlagEnabled([]string{"--sync=false", "--sync"}, "--sync")) + require.False(t, boolFlagEnabled([]string{"--other"}, "--sync")) } func TestIsDiscordID(t *testing.T) { diff --git a/internal/cli/sync_lock.go b/internal/cli/sync_lock.go index 44fa5f50..cccfb027 100644 --- a/internal/cli/sync_lock.go +++ b/internal/cli/sync_lock.go @@ -2,9 +2,13 @@ package cli import ( "context" + "crypto/rand" + "encoding/hex" + "errors" "fmt" "os" "path/filepath" + "strconv" "strings" "time" @@ -12,6 +16,10 @@ import ( ) func (r *runtime) withSyncLock(fn func() error) error { + return r.withSyncLockOperation("writer", fn) +} + +func (r *runtime) withSyncLockOperation(operation string, fn func() error) error { if r.dbLockHeld { return fn() } @@ -19,19 +27,57 @@ func (r *runtime) withSyncLock(fn func() error) error { if err != nil { return err } - release, err := acquireSyncLock(r.ctx, lockPath) + started := r.nowUTC() + token := newSyncLockToken() + release, err := acquireSyncLockWithMetadata(r.ctx, lockPath, syncLockMetadataBody(operation, "locked", started, r.nowUTC(), token)) if err != nil { return err } - r.dbLockHeld = true - r.lockStarted = r.nowUTC() - r.setSyncLockPhase("locked") - defer func() { - r.dbLockHeld = false - r.lockStarted = time.Time{} - _ = release() - }() - return fn() + return r.runWithHeldSyncLock(lockPath, release, operation, started, token, fn) +} + +func (r *runtime) withMessagesSyncLock(fn func() error) error { + if r.dbLockHeld { + return fn() + } + lockPath, err := r.syncLockPath() + if err != nil { + return err + } + ticker := time.NewTicker(50 * time.Millisecond) + defer ticker.Stop() + for { + started := r.nowUTC() + token := newSyncLockToken() + release, locked, err := tryAcquireSyncLockWithMetadata(lockPath, syncLockMetadataBody("messages-sync", "locked", started, r.nowUTC(), token)) + if err != nil { + return err + } + if locked { + return r.runWithHeldSyncLock(lockPath, release, "messages-sync", started, token, fn) + } + // This check runs only after a failed nonblocking lock attempt, so stale + // metadata or PID reuse cannot identify an active tail owner by itself. + if r.activeTailOwnsSyncLock(lockPath) { + started = r.nowUTC() + token = newSyncLockToken() + release, locked, err = tryAcquireSyncLockWithMetadata(lockPath, syncLockMetadataBody("messages-sync", "locked", started, r.nowUTC(), token)) + if err != nil { + return err + } + if locked { + return r.runWithHeldSyncLock(lockPath, release, "messages-sync", started, token, fn) + } + if r.activeTailOwnsSyncLock(lockPath) { + return usageErr(errors.New("tail already owns live sync; omit --sync while tail is running")) + } + } + select { + case <-r.ctx.Done(): + return syncLockErr(r.ctx, lockPath) + case <-ticker.C: + } + } } func (r *runtime) tryWithSyncLock(fn func() error) (bool, error) { @@ -42,19 +88,72 @@ func (r *runtime) tryWithSyncLock(fn func() error) (bool, error) { if err != nil { return false, err } - release, locked, err := tryAcquireSyncLock(lockPath) + started := r.nowUTC() + token := newSyncLockToken() + release, locked, err := tryAcquireSyncLockWithMetadata(lockPath, syncLockMetadataBody("writer", "locked", started, r.nowUTC(), token)) if err != nil || !locked { return locked, err } + return true, r.runWithHeldSyncLock(lockPath, release, "writer", started, token, fn) +} + +func (r *runtime) runWithHeldSyncLock(lockPath string, release func() error, operation string, started time.Time, token string, fn func() error) error { + if strings.TrimSpace(operation) == "tail" && token != "" { + var err error + r.lockTokenFree, err = acquireSyncLockWithMetadata(r.ctx, syncLockTokenPath(lockPath, token), syncLockMetadataBody("tail-token", "locked", started, r.nowUTC(), token)) + if err != nil { + _ = release() + return err + } + } r.dbLockHeld = true - r.lockStarted = r.nowUTC() - r.setSyncLockPhase("locked") + r.lockStarted = started + r.lockOperation = strings.TrimSpace(operation) + r.lockToken = token defer func() { r.dbLockHeld = false r.lockStarted = time.Time{} + r.lockOperation = "" + cleanupToken := r.lockToken + r.lockToken = "" + if r.lockTokenFree != nil { + _ = r.lockTokenFree() + r.lockTokenFree = nil + _ = os.Remove(syncLockTokenPath(lockPath, cleanupToken)) + _ = os.Remove(syncLockMetadataPath(syncLockTokenPath(lockPath, cleanupToken))) + } _ = release() }() - return true, fn() + return fn() +} + +func (r *runtime) activateTailSyncLock() error { + if !r.dbLockHeld { + return nil + } + lockPath, err := r.syncLockPath() + if err != nil { + return err + } + token := r.lockToken + if token == "" { + token = newSyncLockToken() + r.lockToken = token + } + if r.lockTokenFree == nil { + release, err := acquireSyncLockWithMetadata(r.ctx, syncLockTokenPath(lockPath, token), syncLockMetadataBody("tail-token", "locked", r.lockStarted, r.nowUTC(), token)) + if err != nil { + return err + } + r.lockTokenFree = release + } + r.lockOperation = "tail" + started := r.lockStarted + if started.IsZero() { + started = r.nowUTC() + r.lockStarted = started + } + return writeSyncLockMetadataSidecar(lockPath, []byte(syncLockMetadataBody("tail", "live", started, r.nowUTC(), token))) } func (r *runtime) setSyncLockPhase(phase string) { @@ -69,13 +168,8 @@ func (r *runtime) setSyncLockPhase(phase string) { if started.IsZero() { started = r.nowUTC() } - body := fmt.Sprintf("pid=%d\nstarted_at=%s\nupdated_at=%s\nphase=%s\n", - os.Getpid(), - started.Format(time.RFC3339Nano), - r.nowUTC().Format(time.RFC3339Nano), - phase, - ) - _ = os.WriteFile(path, []byte(body), 0o600) + body := syncLockMetadataBody(r.lockOperation, phase, started, r.nowUTC(), r.lockToken) + _ = writeSyncLockMetadataSidecar(path, []byte(body)) } func (r *runtime) syncLockPath() (string, error) { @@ -86,9 +180,155 @@ func (r *runtime) syncLockPath() (string, error) { return filepath.Join(filepath.Dir(dbPath), ".discrawl-sync.lock"), nil } +type syncLockOwner struct { + PID int + Operation string + Token string +} + +func (r *runtime) activeTailOwnsSyncLock(path string) bool { + owner, ok := readSyncLockOwner(path) + if !ok || owner.Operation != "tail" || owner.PID <= 0 || !validSyncLockToken(owner.Token) { + return false + } + if !syncLockPIDAlive(owner.PID) { + return false + } + select { + case <-r.ctx.Done(): + return false + case <-time.After(20 * time.Millisecond): + } + current, ok := readSyncLockOwner(path) + return ok && + current.PID == owner.PID && + current.Operation == owner.Operation && + current.Token == owner.Token && + validSyncLockToken(current.Token) && + syncLockPIDAlive(current.PID) && + syncLockTokenHeld(path, current.Token) +} + +func readSyncLockOwner(path string) (syncLockOwner, bool) { + if owner, ok := readSyncLockOwnerFile(syncLockMetadataPath(path)); ok { + return owner, true + } + return readSyncLockOwnerFile(path) +} + +func readSyncLockOwnerFile(path string) (syncLockOwner, bool) { + body, err := os.ReadFile(path) + if err != nil { + return syncLockOwner{}, false + } + fields := map[string]string{} + for line := range strings.SplitSeq(string(body), "\n") { + line = strings.TrimSpace(line) + if line == "" { + continue + } + key, value, ok := strings.Cut(line, "=") + if !ok { + return syncLockOwner{}, false + } + fields[strings.TrimSpace(key)] = strings.TrimSpace(value) + } + pidRaw := fields["pid"] + if pidRaw == "" { + return syncLockOwner{}, false + } + pid, err := strconv.Atoi(pidRaw) + if err != nil { + return syncLockOwner{}, false + } + return syncLockOwner{PID: pid, Operation: fields["operation"], Token: fields["token"]}, true +} + +func writeSyncLockMetadataRecord(file *os.File, path string, metadata []byte) error { + if _, err := file.Seek(0, 0); err != nil { + return err + } + if err := file.Truncate(0); err != nil { + return err + } + if _, err := file.Write(metadata); err != nil { + return err + } + return writeSyncLockMetadataSidecar(path, metadata) +} + +func writeSyncLockMetadataFiles(path string, metadata []byte) error { + if err := os.WriteFile(path, metadata, 0o600); err != nil { + return err + } + return writeSyncLockMetadataSidecar(path, metadata) +} + +func writeSyncLockMetadataSidecar(path string, metadata []byte) error { + return os.WriteFile(syncLockMetadataPath(path), metadata, 0o600) +} + +func syncLockMetadataPath(lockPath string) string { + return lockPath + ".meta" +} + +func syncLockMetadataBody(operation, phase string, started, updated time.Time, token string) string { + return fmt.Sprintf("pid=%d\noperation=%s\ntoken=%s\nstarted_at=%s\nupdated_at=%s\nphase=%s\n", + os.Getpid(), + strings.TrimSpace(operation), + token, + started.Format(time.RFC3339Nano), + updated.Format(time.RFC3339Nano), + phase, + ) +} + +func newSyncLockToken() string { + var raw [16]byte + if _, err := rand.Read(raw[:]); err != nil { + return fmt.Sprintf("%016x%016x", os.Getpid(), time.Now().UnixNano()) + } + return hex.EncodeToString(raw[:]) +} + +func syncLockTokenPath(lockPath, token string) string { + return lockPath + "." + token + ".owner" +} + +func validSyncLockToken(token string) bool { + if len(token) != 32 { + return false + } + for _, ch := range token { + if (ch < '0' || ch > '9') && (ch < 'a' || ch > 'f') { + return false + } + } + return true +} + +func syncLockTokenHeld(lockPath, token string) bool { + if !validSyncLockToken(token) { + return false + } + tokenPath := syncLockTokenPath(lockPath, token) + if _, err := os.Stat(tokenPath); err != nil { + return false + } + release, locked, err := tryAcquireSyncLock(tokenPath) + if err != nil { + return false + } + if locked { + _ = release() + return false + } + return true +} + func syncLockErr(ctx context.Context, path string) error { if ctx.Err() != nil { - if body, err := os.ReadFile(path); err == nil { + if body, err := readSyncLockMetadata(path); err == nil { details := strings.TrimSpace(string(body)) if details != "" { return fmt.Errorf("wait for sync lock %s (%s): %w", path, strings.ReplaceAll(details, "\n", ", "), ctx.Err()) @@ -98,3 +338,11 @@ func syncLockErr(ctx context.Context, path string) error { } return nil } + +func readSyncLockMetadata(path string) ([]byte, error) { + body, err := os.ReadFile(syncLockMetadataPath(path)) + if err == nil { + return body, nil + } + return os.ReadFile(path) +} diff --git a/internal/cli/sync_lock_other.go b/internal/cli/sync_lock_other.go index e95878b8..989ddad7 100644 --- a/internal/cli/sync_lock_other.go +++ b/internal/cli/sync_lock_other.go @@ -8,6 +8,14 @@ func acquireSyncLock(context.Context, string) (func() error, error) { return func() error { return nil }, nil } +func acquireSyncLockWithMetadata(context.Context, string, string) (func() error, error) { + return func() error { return nil }, nil +} + func tryAcquireSyncLock(string) (func() error, bool, error) { return func() error { return nil }, true, nil } + +func tryAcquireSyncLockWithMetadata(string, string) (func() error, bool, error) { + return func() error { return nil }, true, nil +} diff --git a/internal/cli/sync_lock_pid_other.go b/internal/cli/sync_lock_pid_other.go new file mode 100644 index 00000000..00cbdafc --- /dev/null +++ b/internal/cli/sync_lock_pid_other.go @@ -0,0 +1,7 @@ +//go:build !unix && !windows + +package cli + +func syncLockPIDAlive(pid int) bool { + return pid > 0 +} diff --git a/internal/cli/sync_lock_pid_unix.go b/internal/cli/sync_lock_pid_unix.go new file mode 100644 index 00000000..01307c09 --- /dev/null +++ b/internal/cli/sync_lock_pid_unix.go @@ -0,0 +1,13 @@ +//go:build unix + +package cli + +import "golang.org/x/sys/unix" + +func syncLockPIDAlive(pid int) bool { + if pid <= 0 { + return false + } + err := unix.Kill(pid, 0) + return err == nil || err == unix.EPERM +} diff --git a/internal/cli/sync_lock_pid_windows.go b/internal/cli/sync_lock_pid_windows.go new file mode 100644 index 00000000..b77104a9 --- /dev/null +++ b/internal/cli/sync_lock_pid_windows.go @@ -0,0 +1,17 @@ +//go:build windows + +package cli + +import "golang.org/x/sys/windows" + +func syncLockPIDAlive(pid int) bool { + if pid <= 0 { + return false + } + handle, err := windows.OpenProcess(windows.PROCESS_QUERY_LIMITED_INFORMATION, false, uint32(pid)) + if err != nil { + return false + } + _ = windows.CloseHandle(handle) + return true +} diff --git a/internal/cli/sync_lock_unix.go b/internal/cli/sync_lock_unix.go index eda26fa4..638b48b3 100644 --- a/internal/cli/sync_lock_unix.go +++ b/internal/cli/sync_lock_unix.go @@ -13,6 +13,10 @@ import ( ) func acquireSyncLock(ctx context.Context, path string) (func() error, error) { + return acquireSyncLockWithMetadata(ctx, path, fmt.Sprintf("pid=%d\n", os.Getpid())) +} + +func acquireSyncLockWithMetadata(ctx context.Context, path string, metadata string) (func() error, error) { file, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600) if err != nil { return nil, fmt.Errorf("open sync lock: %w", err) @@ -29,9 +33,11 @@ func acquireSyncLock(ctx context.Context, path string) (func() error, error) { err = unix.Flock(int(file.Fd()), unix.LOCK_EX|unix.LOCK_NB) if err == nil { locked = true - _, _ = file.Seek(0, 0) - _ = file.Truncate(0) - _, _ = fmt.Fprintf(file, "pid=%d\n", os.Getpid()) + if err := writeSyncLockMetadataRecord(file, path, []byte(metadata)); err != nil { + _ = unix.Flock(int(file.Fd()), unix.LOCK_UN) + locked = false + return nil, fmt.Errorf("write sync lock metadata: %w", err) + } return func() error { unlockErr := unix.Flock(int(file.Fd()), unix.LOCK_UN) closeErr := file.Close() @@ -53,6 +59,10 @@ func acquireSyncLock(ctx context.Context, path string) (func() error, error) { } func tryAcquireSyncLock(path string) (func() error, bool, error) { + return tryAcquireSyncLockWithMetadata(path, fmt.Sprintf("pid=%d\n", os.Getpid())) +} + +func tryAcquireSyncLockWithMetadata(path string, metadata string) (func() error, bool, error) { file, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600) if err != nil { return nil, false, fmt.Errorf("open sync lock: %w", err) @@ -65,9 +75,11 @@ func tryAcquireSyncLock(path string) (func() error, bool, error) { } return nil, false, fmt.Errorf("acquire sync lock: %w", err) } - _, _ = file.Seek(0, 0) - _ = file.Truncate(0) - _, _ = fmt.Fprintf(file, "pid=%d\n", os.Getpid()) + if err := writeSyncLockMetadataRecord(file, path, []byte(metadata)); err != nil { + _ = unix.Flock(int(file.Fd()), unix.LOCK_UN) + _ = file.Close() + return nil, false, fmt.Errorf("write sync lock metadata: %w", err) + } return func() error { unlockErr := unix.Flock(int(file.Fd()), unix.LOCK_UN) closeErr := file.Close() diff --git a/internal/cli/sync_lock_windows.go b/internal/cli/sync_lock_windows.go index cb3cdd1d..b8edfb89 100644 --- a/internal/cli/sync_lock_windows.go +++ b/internal/cli/sync_lock_windows.go @@ -12,6 +12,10 @@ import ( ) func acquireSyncLock(ctx context.Context, path string) (func() error, error) { + return acquireSyncLockWithMetadata(ctx, path, fmt.Sprintf("pid=%d\n", os.Getpid())) +} + +func acquireSyncLockWithMetadata(ctx context.Context, path string, metadata string) (func() error, error) { file, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600) if err != nil { return nil, fmt.Errorf("open sync lock: %w", err) @@ -25,14 +29,16 @@ func acquireSyncLock(ctx context.Context, path string) (func() error, error) { ticker := time.NewTicker(200 * time.Millisecond) defer ticker.Stop() handle := windows.Handle(file.Fd()) - overlapped := &windows.Overlapped{} + overlapped := syncLockWindowsOverlapped() for { err = windows.LockFileEx(handle, windows.LOCKFILE_EXCLUSIVE_LOCK|windows.LOCKFILE_FAIL_IMMEDIATELY, 0, 1, 0, overlapped) if err == nil { locked = true - _, _ = file.Seek(0, 0) - _ = file.Truncate(0) - _, _ = fmt.Fprintf(file, "pid=%d\n", os.Getpid()) + if err := writeSyncLockMetadataRecord(file, path, []byte(metadata)); err != nil { + _ = windows.UnlockFileEx(handle, 0, 1, 0, overlapped) + locked = false + return nil, fmt.Errorf("write sync lock metadata: %w", err) + } return func() error { unlockErr := windows.UnlockFileEx(handle, 0, 1, 0, overlapped) closeErr := file.Close() @@ -51,20 +57,26 @@ func acquireSyncLock(ctx context.Context, path string) (func() error, error) { } func tryAcquireSyncLock(path string) (func() error, bool, error) { + return tryAcquireSyncLockWithMetadata(path, fmt.Sprintf("pid=%d\n", os.Getpid())) +} + +func tryAcquireSyncLockWithMetadata(path string, metadata string) (func() error, bool, error) { file, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600) if err != nil { return nil, false, fmt.Errorf("open sync lock: %w", err) } handle := windows.Handle(file.Fd()) - overlapped := &windows.Overlapped{} + overlapped := syncLockWindowsOverlapped() err = windows.LockFileEx(handle, windows.LOCKFILE_EXCLUSIVE_LOCK|windows.LOCKFILE_FAIL_IMMEDIATELY, 0, 1, 0, overlapped) if err != nil { _ = file.Close() return nil, false, nil } - _, _ = file.Seek(0, 0) - _ = file.Truncate(0) - _, _ = fmt.Fprintf(file, "pid=%d\n", os.Getpid()) + if err := writeSyncLockMetadataRecord(file, path, []byte(metadata)); err != nil { + _ = windows.UnlockFileEx(handle, 0, 1, 0, overlapped) + _ = file.Close() + return nil, false, fmt.Errorf("write sync lock metadata: %w", err) + } return func() error { unlockErr := windows.UnlockFileEx(handle, 0, 1, 0, overlapped) closeErr := file.Close() @@ -74,3 +86,7 @@ func tryAcquireSyncLock(path string) (func() error, bool, error) { return closeErr }, true, nil } + +func syncLockWindowsOverlapped() *windows.Overlapped { + return &windows.Overlapped{} +} diff --git a/internal/cli/sync_lock_windows_test.go b/internal/cli/sync_lock_windows_test.go new file mode 100644 index 00000000..e654114a --- /dev/null +++ b/internal/cli/sync_lock_windows_test.go @@ -0,0 +1,12 @@ +//go:build windows + +package cli + +import "testing" + +func TestSyncLockWindowsOverlappedKeepsByteZeroCompatibility(t *testing.T) { + overlapped := syncLockWindowsOverlapped() + if overlapped.Offset != 0 || overlapped.OffsetHigh != 0 { + t.Fatal("windows sync lock must keep byte 0 compatibility with older binaries") + } +} diff --git a/internal/discord/client.go b/internal/discord/client.go index d35914a6..12a08b91 100644 --- a/internal/discord/client.go +++ b/internal/discord/client.go @@ -21,6 +21,10 @@ type EventHandler interface { OnMemberDelete(context.Context, string, string) error } +type TailReadyHandler interface { + OnTailReady(context.Context) error +} + type Client struct { session *discordgo.Session requestTimeout time.Duration @@ -278,6 +282,11 @@ func (c *Client) Tail(ctx context.Context, handler EventHandler) error { _ = c.session.Close() wg.Wait() }() + if ready, ok := handler.(TailReadyHandler); ok { + if err := ready.OnTailReady(tailCtx); err != nil { + return err + } + } select { case <-ctx.Done(): return nil diff --git a/internal/discord/client_test.go b/internal/discord/client_test.go index 32f181e3..29d95a2f 100644 --- a/internal/discord/client_test.go +++ b/internal/discord/client_test.go @@ -353,6 +353,7 @@ func TestTailReceivesGatewayEvents(t *testing.T) { require.Equal(t, 1, handler.channels) require.Equal(t, 1, handler.memberUpserts) require.Equal(t, 1, handler.memberDeletes) + require.Equal(t, 1, handler.ready) } func TestTailFailsFastWhenWorkerQueueFills(t *testing.T) { @@ -510,6 +511,12 @@ type recordingHandler struct { channels int memberUpserts int memberDeletes int + ready int +} + +func (r *recordingHandler) OnTailReady(context.Context) error { + r.ready++ + return nil } func (r *recordingHandler) OnMessageCreate(context.Context, *discordgo.Message) error { diff --git a/internal/syncer/syncer.go b/internal/syncer/syncer.go index b33c5fcd..29f46892 100644 --- a/internal/syncer/syncer.go +++ b/internal/syncer/syncer.go @@ -27,6 +27,10 @@ type Client interface { Tail(context.Context, discordclient.EventHandler) error } +type closeableClient interface { + Close() error +} + type Syncer struct { client Client store *store.Store @@ -37,6 +41,7 @@ type Syncer struct { messageChannelTimeout time.Duration messageSyncLogEvery time.Duration messageSyncWaitEvery time.Duration + tailReady func(context.Context) error } type SyncOptions struct { @@ -51,6 +56,10 @@ type SyncOptions struct { RepairReason string } +func (s *Syncer) SetTailReadyCallback(fn func(context.Context) error) { + s.tailReady = fn +} + type SyncStats struct { Guilds int `json:"guilds"` Channels int `json:"channels"` diff --git a/internal/syncer/syncer_tail_test.go b/internal/syncer/syncer_tail_test.go index fe2f7993..622cbd98 100644 --- a/internal/syncer/syncer_tail_test.go +++ b/internal/syncer/syncer_tail_test.go @@ -10,6 +10,7 @@ import ( "github.com/bwmarrin/discordgo" "github.com/stretchr/testify/require" + discordclient "github.com/openclaw/discrawl/internal/discord" "github.com/openclaw/discrawl/internal/store" ) @@ -276,3 +277,77 @@ func TestRunTailWithRepairLoop(t *testing.T) { require.NoError(t, err) require.GreaterOrEqual(t, status.MessageCount, 1) } + +func TestRunTailWithRepairLoopJoinsTailOnCancellation(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + + s, err := store.Open(ctx, filepath.Join(t.TempDir(), "discrawl.db")) + require.NoError(t, err) + defer func() { _ = s.Close() }() + + client := &joiningTailClient{ + started: make(chan struct{}), + finished: make(chan struct{}), + closed: make(chan struct{}), + } + svc := New(client, s, nil) + done := make(chan error, 1) + go func() { + done <- svc.RunTail(ctx, nil, time.Hour) + }() + + select { + case <-client.started: + case <-time.After(time.Second): + t.Fatal("tail did not start") + } + cancel() + require.NoError(t, <-done) + select { + case <-client.finished: + default: + t.Fatal("RunTail returned before client.Tail finished") + } + select { + case <-client.closed: + default: + t.Fatal("RunTail returned before closing tail client") + } +} + +func TestTailReadyCallback(t *testing.T) { + t.Parallel() + + called := false + svc := New(&fakeClient{}, nil, nil) + svc.SetTailReadyCallback(func(context.Context) error { + called = true + return nil + }) + handler := &tailHandler{onReady: svc.tailReady} + require.NoError(t, handler.OnTailReady(context.Background())) + require.True(t, called) + + handler.onReady = nil + require.NoError(t, handler.OnTailReady(context.Background())) +} + +type joiningTailClient struct { + fakeClient + started chan struct{} + finished chan struct{} + closed chan struct{} +} + +func (c *joiningTailClient) Tail(ctx context.Context, _ discordclient.EventHandler) error { + close(c.started) + <-ctx.Done() + close(c.finished) + return nil +} + +func (c *joiningTailClient) Close() error { + close(c.closed) + return nil +} diff --git a/internal/syncer/tail.go b/internal/syncer/tail.go index eb661a59..40c8839a 100644 --- a/internal/syncer/tail.go +++ b/internal/syncer/tail.go @@ -14,19 +14,27 @@ func (s *Syncer) RunTail(ctx context.Context, guildIDs []string, repairEvery tim store: s.store, client: s.client, attachmentTextEnabled: s.attachmentTextEnabled, + onReady: s.tailReady, } if repairEvery <= 0 { return s.client.Tail(ctx, handler) } + tailCtx, cancelTail := context.WithCancel(ctx) + defer cancelTail() errCh := make(chan error, 2) go func() { - errCh <- s.client.Tail(ctx, handler) + errCh <- s.client.Tail(tailCtx, handler) }() ticker := time.NewTicker(repairEvery) defer ticker.Stop() for { select { case <-ctx.Done(): + cancelTail() + if closeable, ok := s.client.(closeableClient); ok { + _ = closeable.Close() + } + <-errCh return nil case err := <-errCh: return err @@ -43,6 +51,14 @@ type tailHandler struct { store *store.Store client Client attachmentTextEnabled bool + onReady func(context.Context) error +} + +func (t *tailHandler) OnTailReady(ctx context.Context) error { + if t.onReady == nil { + return nil + } + return t.onReady(ctx) } func (t *tailHandler) OnMessageCreate(ctx context.Context, msg *discordgo.Message) error {