fix(syncer): reject non-advancing message page cursors

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
 

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.

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
 }

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)
+			})
+		}
+	}
+}

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

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,