diff --git a/CHANGELOG.md b/CHANGELOG.md index 41aa604..fac5808 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ ### Fixes - Stop guild and archived-thread pagination with a cursor error when Discord repeats a page instead of hanging sync. Thanks @SebTardif. +- Reject repeated or missing message-page cursors without losing the last usable backfill checkpoint. Thanks @SebTardif. ### Maintenance diff --git a/docs/commands/sync.md b/docs/commands/sync.md index a2f67f2..fa9820e 100644 --- a/docs/commands/sync.md +++ b/docs/commands/sync.md @@ -77,6 +77,7 @@ discrawl sync --with-media - Every run ends with a `message sync finished` summary. - Each channel crawl has a bounded runtime budget; pathological channels are deferred and retried next sync. - Guild and archived-thread pagination reports a cursor error if Discord repeats a page instead of continuing indefinitely. +- Full message pages with missing or repeated cursors stop the channel crawl with an error and preserve its last usable backfill checkpoint. - Retryable failures and unavailable-channel markers are tracked per channel; stale unavailable markers are cleared after a later successful crawl. - Marker cleanup is best-effort, so one missing local sync-state row cannot crash the run. - Member refresh is best-effort and gives up after five minutes without a caller-supplied deadline. Routine latest-only syncs skip it unless `--with-members` is set. diff --git a/internal/syncer/message_sync.go b/internal/syncer/message_sync.go index 84e7784..0154ff3 100644 --- a/internal/syncer/message_sync.go +++ b/internal/syncer/message_sync.go @@ -417,7 +417,6 @@ func (s *Syncer) bootstrapChannelHistory(ctx context.Context, channel *discordgo } break } - before = page[len(page)-1].ID if len(page) < 100 { if newest != "" { if err := s.store.SetSyncState(ctx, channelHistoryCompleteScope(channel.ID), "1"); err != nil { @@ -426,6 +425,14 @@ func (s *Syncer) bootstrapChannelHistory(ctx context.Context, channel *discordgo } break } + nextBefore := page[len(page)-1].ID + if nextBefore == "" { + return messageCount, fmt.Errorf("channel %s message page missing id", channel.ID) + } + if nextBefore == before { + return messageCount, fmt.Errorf("channel %s message page cursor did not advance", channel.ID) + } + before = nextBefore } if newest != "" { if err := s.advanceChannelLatest(ctx, channel.ID, newest); err != nil { @@ -451,7 +458,6 @@ func (s *Syncer) syncForwardPages(ctx context.Context, channel *discordgo.Channe return messageCount, newest, err } progress.touch(channel, len(page)) - after = maxSnowflake(after, pageNewest) newest = maxSnowflake(newest, pageNewest) messageCount += len(page) if err := s.advanceChannelLatest(ctx, channel.ID, newest); err != nil { @@ -460,6 +466,14 @@ func (s *Syncer) syncForwardPages(ctx context.Context, channel *discordgo.Channe if len(page) < 100 { break } + nextAfter := maxSnowflake(after, pageNewest) + if nextAfter == "" { + return messageCount, newest, fmt.Errorf("channel %s message page missing id", channel.ID) + } + if nextAfter == after { + return messageCount, newest, fmt.Errorf("channel %s message page cursor did not advance", channel.ID) + } + after = nextAfter } return messageCount, newest, nil } @@ -507,8 +521,17 @@ func (s *Syncer) syncBackfillPages(ctx context.Context, channel *discordgo.Chann } break } - before = page[len(page)-1].ID - if err := s.store.SetSyncState(ctx, channelBackfillScope(channel.ID), before); err != nil { + nextBefore := page[len(page)-1].ID + // Even a bounded pass must leave a usable checkpoint for its next run. + if len(page) == 100 { + if nextBefore == "" { + return messageCount, newest, fmt.Errorf("channel %s message page missing id", channel.ID) + } + if nextBefore == before { + return messageCount, newest, fmt.Errorf("channel %s message page cursor did not advance", channel.ID) + } + } + if err := s.store.SetSyncState(ctx, channelBackfillScope(channel.ID), nextBefore); err != nil { return messageCount, newest, err } if len(page) < 100 { @@ -520,6 +543,7 @@ func (s *Syncer) syncBackfillPages(ctx context.Context, channel *discordgo.Chann if pageLimit > 0 && pages >= pageLimit { break } + before = nextBefore } return messageCount, newest, nil } diff --git a/internal/syncer/message_sync_cursor_test.go b/internal/syncer/message_sync_cursor_test.go new file mode 100644 index 0000000..dd63b67 --- /dev/null +++ b/internal/syncer/message_sync_cursor_test.go @@ -0,0 +1,191 @@ +package syncer + +import ( + "context" + "fmt" + "path/filepath" + "strconv" + "testing" + "time" + + "github.com/bwmarrin/discordgo" + "github.com/stretchr/testify/require" + + "github.com/openclaw/discrawl/internal/store" +) + +type repeatingMessagePageClient struct { + fakeClient + page []*discordgo.Message + requests int +} + +func (c *repeatingMessagePageClient) ChannelMessages(_ context.Context, _ string, _ int, _, _ string) ([]*discordgo.Message, error) { + c.requests++ + if c.requests > 5 { + return nil, fmt.Errorf("message pager called %d times without stopping", c.requests) + } + return c.page, nil +} + +func fullMessagePage(lastID string) []*discordgo.Message { + page := make([]*discordgo.Message, 100) + now := time.Now().UTC() + author := &discordgo.User{ID: "u1", Username: "user"} + for i := range 99 { + page[i] = &discordgo.Message{ + ID: strconv.Itoa(i + 1), + GuildID: "g1", + ChannelID: "c1", + Content: "msg", + Timestamp: now, + Author: author, + } + } + page[99] = &discordgo.Message{ + ID: lastID, + GuildID: "g1", + ChannelID: "c1", + Content: "tail", + Timestamp: now, + Author: author, + } + return page +} + +func emptyIDMessagePage() []*discordgo.Message { + page := make([]*discordgo.Message, 100) + now := time.Now().UTC() + author := &discordgo.User{ID: "u1", Username: "user"} + for i := range page { + page[i] = &discordgo.Message{ + ID: "", + GuildID: "g1", + ChannelID: "c1", + Content: "msg", + Timestamp: now, + Author: author, + } + } + return page +} + +func TestMessagePagesErrorWhenCursorDoesNotAdvance(t *testing.T) { + t.Parallel() + + channel := &discordgo.Channel{ID: "c1", GuildID: "g1", Name: "general", Type: discordgo.ChannelTypeGuildText} + + tests := []struct { + name string + page []*discordgo.Message + run func(*Syncer, *discordgo.Channel) error + wantErr string + wantCalls int + }{ + { + name: "bootstrap repeats last id", + page: fullMessagePage("100"), + run: func(svc *Syncer, channel *discordgo.Channel) error { + _, err := svc.bootstrapChannelHistory(context.Background(), channel, false, time.Time{}, nil) + return err + }, + wantErr: "message page cursor did not advance", + wantCalls: 2, + }, + { + name: "bootstrap empty last id", + page: fullMessagePage(""), + run: func(svc *Syncer, channel *discordgo.Channel) error { + _, err := svc.bootstrapChannelHistory(context.Background(), channel, false, time.Time{}, nil) + return err + }, + wantErr: "message page missing id", + wantCalls: 1, + }, + { + name: "forward repeats newest id", + page: fullMessagePage("100"), + run: func(svc *Syncer, channel *discordgo.Channel) error { + _, _, err := svc.syncForwardPages(context.Background(), channel, "50", false, nil) + return err + }, + wantErr: "message page cursor did not advance", + wantCalls: 2, + }, + { + name: "forward empty newest id", + page: emptyIDMessagePage(), + run: func(svc *Syncer, channel *discordgo.Channel) error { + _, _, err := svc.syncForwardPages(context.Background(), channel, "100", false, nil) + return err + }, + wantErr: "message page cursor did not advance", + wantCalls: 1, + }, + { + name: "unlimited backfill repeats last id", + page: fullMessagePage("100"), + run: func(svc *Syncer, channel *discordgo.Channel) error { + _, _, err := svc.syncBackfillPages(context.Background(), channel, "", "", channel.Name, false, time.Time{}, 0, nil) + return err + }, + wantErr: "message page cursor did not advance", + wantCalls: 2, + }, + { + name: "unlimited backfill empty last id", + page: fullMessagePage(""), + run: func(svc *Syncer, channel *discordgo.Channel) error { + _, _, err := svc.syncBackfillPages(context.Background(), channel, "", "", channel.Name, false, time.Time{}, 0, nil) + return err + }, + wantErr: "message page missing id", + wantCalls: 1, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + ctx := context.Background() + s, err := store.Open(ctx, filepath.Join(t.TempDir(), "discrawl.db")) + require.NoError(t, err) + t.Cleanup(func() { _ = s.Close() }) + + client := &repeatingMessagePageClient{page: tt.page} + svc := New(client, s, nil) + err = tt.run(svc, channel) + require.ErrorContains(t, err, tt.wantErr) + require.Equal(t, tt.wantCalls, client.requests) + }) + } +} + +func TestInvalidBackfillPagePreservesCheckpoint(t *testing.T) { + t.Parallel() + for _, pageLimit := range []int{0, 1} { + for _, lastID := range []string{"", "100"} { + t.Run(fmt.Sprintf("limit=%d/last=%q", pageLimit, lastID), func(t *testing.T) { + t.Parallel() + ctx := t.Context() + st, err := store.Open(ctx, filepath.Join(t.TempDir(), "archive.db")) + require.NoError(t, err) + t.Cleanup(func() { _ = st.Close() }) + require.NoError(t, st.SetSyncState(ctx, channelBackfillScope("c1"), "100")) + client := &repeatingMessagePageClient{page: fullMessagePage(lastID)} + svc := New(client, st, nil) + channel := &discordgo.Channel{ID: "c1", GuildID: "g1", Name: "general"} + _, _, err = svc.syncBackfillPages(ctx, channel, "100", "200", channel.Name, false, time.Time{}, pageLimit, nil) + require.Error(t, err) + require.Equal(t, 1, client.requests) + cursor, err := st.GetSyncState(ctx, channelBackfillScope("c1")) + require.NoError(t, err) + require.Equal(t, "100", cursor) + complete, err := st.GetSyncState(ctx, channelHistoryCompleteScope("c1")) + require.NoError(t, err) + require.Empty(t, complete) + }) + } + } +} diff --git a/internal/syncer/message_sync_helpers_test.go b/internal/syncer/message_sync_helpers_test.go index a02f6f5..681ca26 100644 --- a/internal/syncer/message_sync_helpers_test.go +++ b/internal/syncer/message_sync_helpers_test.go @@ -200,8 +200,7 @@ func TestMessageChannelConcurrentErrorAndProgressBranches(t *testing.T) { func TestMessageChannelConcurrentFatalErrorCancelsPeers(t *testing.T) { t.Parallel() - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() + ctx := t.Context() s, err := store.Open(ctx, filepath.Join(t.TempDir(), "discrawl.db")) require.NoError(t, err) defer func() { _ = s.Close() }() @@ -220,6 +219,8 @@ func TestMessageChannelConcurrentFatalErrorCancelsPeers(t *testing.T) { } svc := New(client, s, slog.New(slog.DiscardHandler)) + ctx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() done := make(chan struct { count int err error diff --git a/internal/syncer/syncer_tail_test.go b/internal/syncer/syncer_tail_test.go index f557ea1..959391d 100644 --- a/internal/syncer/syncer_tail_test.go +++ b/internal/syncer/syncer_tail_test.go @@ -313,8 +313,7 @@ func TestTailHandlerMessageUpdateFailureUsesSyncerRefetchedMetadata(t *testing.T for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() + ctx := t.Context() var tailStore *store.Store var onFetch func(context.Context) @@ -376,6 +375,8 @@ func TestTailHandlerMessageUpdateFailureUsesSyncerRefetchedMetadata(t *testing.T defer func() { _ = eventClient.Close() }() setDiscordTailHandlerTimeout(t, eventClient, 25*time.Millisecond) + ctx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() handler := &capturingTailHandler{ tailHandler: &tailHandler{ store: tailStore,