diff --git a/internal/store/query.go b/internal/store/query.go index 94013dea..6092cee3 100644 --- a/internal/store/query.go +++ b/internal/store/query.go @@ -59,6 +59,16 @@ func (s *Store) GetSyncState(ctx context.Context, scope string) (string, error) return cursor, nil } +// ChannelHasMessages reports whether any message rows are stored for the +// channel. It is an O(1) existence probe against idx_messages_channel_id and is +// safe to call once per channel per sync. +func (s *Store) ChannelHasMessages(ctx context.Context, channelID string) (bool, error) { + if channelID == "" { + return false, nil + } + return s.q.ChannelHasMessages(ctx, channelID) +} + func (s *Store) ChannelMessageBounds(ctx context.Context, channelID string) (string, string, error) { row, err := s.q.ChannelMessageBounds(ctx, channelID) if err != nil { diff --git a/internal/store/sqlc/queries.sql b/internal/store/sqlc/queries.sql index 0dec26ab..186b74cd 100644 --- a/internal/store/sqlc/queries.sql +++ b/internal/store/sqlc/queries.sql @@ -14,6 +14,9 @@ on conflict(scope) do update set delete from sync_state where scope = ?; +-- name: ChannelHasMessages :one +select exists(select 1 from messages where channel_id = ? limit 1) as has_messages; + -- name: ChannelMessageBounds :one select cast(coalesce(min(id), '') as text) as oldest_id, cast(coalesce(max(id), '') as text) as newest_id diff --git a/internal/store/storedb/queries.sql.go b/internal/store/storedb/queries.sql.go index c9b18f6d..bb537868 100644 --- a/internal/store/storedb/queries.sql.go +++ b/internal/store/storedb/queries.sql.go @@ -54,6 +54,17 @@ func (q *Queries) CatalogIntegrity(ctx context.Context) (CatalogIntegrityRow, er return i, err } +const channelHasMessages = `-- name: ChannelHasMessages :one +select exists(select 1 from messages where channel_id = ? limit 1) as has_messages +` + +func (q *Queries) ChannelHasMessages(ctx context.Context, channelID string) (bool, error) { + row := q.db.QueryRowContext(ctx, channelHasMessages, channelID) + var has_messages bool + err := row.Scan(&has_messages) + return has_messages, err +} + const channelMessageBounds = `-- name: ChannelMessageBounds :one select cast(coalesce(min(id), '') as text) as oldest_id, cast(coalesce(max(id), '') as text) as newest_id diff --git a/internal/syncer/channel_catalog_test.go b/internal/syncer/channel_catalog_test.go index 8d672003..8167a415 100644 --- a/internal/syncer/channel_catalog_test.go +++ b/internal/syncer/channel_catalog_test.go @@ -508,6 +508,13 @@ func TestSyncSkipsUnchangedThreadsWhenHistoryComplete(t *testing.T) { })) require.NoError(t, s.SetSyncState(ctx, channelLatestScope("t1"), "200")) require.NoError(t, s.SetSyncState(ctx, channelHistoryCompleteScope("t1"), "1")) + // A genuinely complete channel has the messages to show for it. Without a + // stored row the channel is re-crawled to recover the missing history. + require.NoError(t, s.UpsertMessage(ctx, store.MessageRecord{ + ID: "200", GuildID: "g1", ChannelID: "t1", ChannelName: "bug-report", + AuthorID: "u1", AuthorName: "User", CreatedAt: time.Now().UTC().Format(time.RFC3339Nano), + Content: "hello", NormalizedContent: "hello", RawJSON: `{}`, + })) client := &fakeClient{ guilds: []*discordgo.UserGuild{{ID: "g1", Name: "Guild"}}, @@ -556,6 +563,13 @@ func TestSyncSkipsUnchangedTextChannelsWhenHistoryComplete(t *testing.T) { })) require.NoError(t, s.SetSyncState(ctx, channelLatestScope("c1"), "200")) require.NoError(t, s.SetSyncState(ctx, channelHistoryCompleteScope("c1"), "1")) + // A genuinely complete channel has the messages to show for it. Without a + // stored row the channel is re-crawled to recover the missing history. + require.NoError(t, s.UpsertMessage(ctx, store.MessageRecord{ + ID: "200", GuildID: "g1", ChannelID: "c1", ChannelName: "general", + AuthorID: "u1", AuthorName: "User", CreatedAt: time.Now().UTC().Format(time.RFC3339Nano), + Content: "hello", NormalizedContent: "hello", RawJSON: `{}`, + })) client := &fakeClient{ guilds: []*discordgo.UserGuild{{ID: "g1", Name: "Guild"}}, diff --git a/internal/syncer/history_verification_test.go b/internal/syncer/history_verification_test.go new file mode 100644 index 00000000..9dab8329 --- /dev/null +++ b/internal/syncer/history_verification_test.go @@ -0,0 +1,532 @@ +package syncer + +import ( + "context" + "errors" + "path/filepath" + "strconv" + "testing" + "time" + + "github.com/bwmarrin/discordgo" + "github.com/stretchr/testify/require" + + "github.com/openclaw/discrawl/internal/store" +) + +func TestNeedsHistoryVerification(t *testing.T) { + t.Parallel() + + channel := &discordgo.Channel{ID: "c1", LastMessageID: "200"} + silent := &discordgo.Channel{ID: "c1"} + windowed := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + + require.False(t, needsHistoryVerification(nil, channelSyncState{BackfillComplete: true}, time.Time{})) + // Not marked complete: the normal backfill path already covers it. + require.False(t, needsHistoryVerification(channel, channelSyncState{}, time.Time{})) + // Marked complete and messages are actually stored: trustworthy. + require.False(t, needsHistoryVerification(channel, channelSyncState{BackfillComplete: true, HasMessages: true}, time.Time{})) + // Marked complete, nothing stored, Discord reports content: verify. + require.True(t, needsHistoryVerification(channel, channelSyncState{BackfillComplete: true}, time.Time{})) + // Already verified empty once: do not crawl again. + require.False(t, needsHistoryVerification(channel, channelSyncState{BackfillComplete: true, VerifiedEmpty: true}, time.Time{})) + // Discord reports no content either: nothing to recover. + require.False(t, needsHistoryVerification(silent, channelSyncState{BackfillComplete: true}, time.Time{})) + // A windowed run cannot complete a recovery, so it does not start one. + require.False(t, needsHistoryVerification(channel, channelSyncState{BackfillComplete: true}, windowed)) + + // shouldSkipChannelSync must defer to it. + require.False(t, shouldSkipChannelSync(channel, channelSyncState{BackfillComplete: true, Latest: "300"}, time.Time{})) + require.True(t, shouldSkipChannelSync(channel, channelSyncState{BackfillComplete: true, Latest: "300", HasMessages: true}, time.Time{})) + require.True(t, shouldSkipChannelSync(channel, channelSyncState{BackfillComplete: true, Latest: "300", VerifiedEmpty: true}, time.Time{})) + // A channel Discord reports as empty still skips on an empty cursor. + require.True(t, shouldSkipChannelSync(silent, channelSyncState{BackfillComplete: true}, time.Time{})) + // Under a window the stranded channel takes its unchanged pre-verification + // path, which for a cursor at the channel head is a skip. + require.True(t, shouldSkipChannelSync(channel, channelSyncState{BackfillComplete: true, Latest: "300"}, windowed)) +} + +func verificationFixture(t *testing.T, messages []*discordgo.Message) (context.Context, *store.Store, *fakeClient, *Syncer, *discordgo.Channel) { + t.Helper() + return verificationFixtureAt(t, messages, "300") +} + +// verificationFixtureAt builds the stranded shape around an explicit channel +// head so a fixture can carry more than one page of history. +func verificationFixtureAt(t *testing.T, messages []*discordgo.Message, head string) (context.Context, *store.Store, *fakeClient, *Syncer, *discordgo.Channel) { + t.Helper() + + ctx := context.Background() + s, err := store.Open(ctx, filepath.Join(t.TempDir(), "discrawl.db")) + require.NoError(t, err) + t.Cleanup(func() { _ = s.Close() }) + require.NoError(t, s.UpsertChannel(ctx, store.ChannelRecord{ID: "c1", GuildID: "g1", Kind: "text", Name: "general", RawJSON: `{}`})) + + // The stranded shape seen in production: history_complete plus a latest + // cursor at or ahead of the channel head, but no message rows at all. + require.NoError(t, s.SetSyncState(ctx, channelHistoryCompleteScope("c1"), "1")) + require.NoError(t, s.SetSyncState(ctx, channelLatestScope("c1"), head)) + + client := &fakeClient{messages: map[string][]*discordgo.Message{"c1": messages}} + channel := &discordgo.Channel{ID: "c1", GuildID: "g1", Name: "general", Type: discordgo.ChannelTypeGuildText, LastMessageID: head} + return ctx, s, client, New(client, s, nil), channel +} + +func storedMessage(id string) *discordgo.Message { + return &discordgo.Message{ + ID: id, + ChannelID: "c1", + GuildID: "g1", + Author: &discordgo.User{ID: "u1", Username: "user"}, + Content: "hello", + Timestamp: time.Date(2026, 5, 8, 12, 0, 0, 0, time.UTC), + } +} + +// storedMessages returns n messages newest first. The ids are equal-width +// decimals so the fake client's string comparisons order them the way Discord +// orders snowflakes, and the store accepts them as canonical. +func storedMessages(n int) []*discordgo.Message { + out := make([]*discordgo.Message, 0, n) + for i := n - 1; i >= 0; i-- { + out = append(out, storedMessage(strconv.Itoa(1000+i))) + } + return out +} + +func TestHistoryVerificationRecoversStrandedChannel(t *testing.T) { + t.Parallel() + + ctx, s, client, svc, channel := verificationFixture(t, []*discordgo.Message{storedMessage("300"), storedMessage("100")}) + + count, err := svc.syncChannelMessages(ctx, "g1", channel, false, false, time.Time{}, false, nil) + require.NoError(t, err) + require.Equal(t, 2, count) + require.Positive(t, client.messageCalls["c1"]) + + has, err := s.ChannelHasMessages(ctx, "c1") + require.NoError(t, err) + require.True(t, has) + + // A recovered channel is not marked verified empty. + marker, err := s.GetSyncState(ctx, channelVerifiedEmptyScope("c1")) + require.NoError(t, err) + require.Empty(t, marker) + + // Second run sees stored messages and skips again. + before := client.messageCalls["c1"] + count, err = svc.syncChannelMessages(ctx, "g1", channel, false, false, time.Time{}, false, nil) + require.NoError(t, err) + require.Zero(t, count) + require.Equal(t, before, client.messageCalls["c1"]) +} + +func TestHistoryVerificationStopsAfterGenuinelyEmptyChannel(t *testing.T) { + t.Parallel() + + ctx, s, client, svc, channel := verificationFixture(t, nil) + + count, err := svc.syncChannelMessages(ctx, "g1", channel, false, false, time.Time{}, false, nil) + require.NoError(t, err) + require.Zero(t, count) + firstRun := client.messageCalls["c1"] + require.Positive(t, firstRun) + + marker, err := s.GetSyncState(ctx, channelVerifiedEmptyScope("c1")) + require.NoError(t, err) + require.Equal(t, "1", marker) + + // Every later run must be a no-op: no re-fetch loop. + for range 3 { + count, err = svc.syncChannelMessages(ctx, "g1", channel, false, false, time.Time{}, false, nil) + require.NoError(t, err) + require.Zero(t, count) + } + require.Equal(t, firstRun, client.messageCalls["c1"]) + + // The channel is still empty, so nothing may retire the marker that stops + // the re-fetch loop. + marker, err = s.GetSyncState(ctx, channelVerifiedEmptyScope("c1")) + require.NoError(t, err) + require.Equal(t, "1", marker) +} + +func TestHistoryVerificationSkipsChannelWithMessages(t *testing.T) { + t.Parallel() + + ctx, s, client, svc, channel := verificationFixture(t, []*discordgo.Message{storedMessage("300")}) + require.NoError(t, s.UpsertMessage(ctx, store.MessageRecord{ + ID: "300", GuildID: "g1", ChannelID: "c1", ChannelName: "general", + AuthorID: "u1", AuthorName: "user", CreatedAt: time.Now().UTC().Format(time.RFC3339Nano), + Content: "hello", NormalizedContent: "hello", RawJSON: `{}`, + })) + + count, err := svc.syncChannelMessages(ctx, "g1", channel, false, false, time.Time{}, false, nil) + require.NoError(t, err) + require.Zero(t, count) + require.Zero(t, client.messageCalls["c1"]) +} + +func TestHistoryVerificationIgnoresChannelWithoutLastMessage(t *testing.T) { + t.Parallel() + + ctx, s, client, svc, _ := verificationFixture(t, nil) + require.NoError(t, s.DeleteSyncState(ctx, channelLatestScope("c1"))) + silent := &discordgo.Channel{ID: "c1", GuildID: "g1", Name: "general", Type: discordgo.ChannelTypeGuildText} + + count, err := svc.syncChannelMessages(ctx, "g1", silent, false, false, time.Time{}, false, nil) + require.NoError(t, err) + require.Zero(t, count) + require.Zero(t, client.messageCalls["c1"]) + + // Nothing was crawled, so nothing may claim to have been verified. + marker, err := s.GetSyncState(ctx, channelVerifiedEmptyScope("c1")) + require.NoError(t, err) + require.Empty(t, marker) +} + +func TestHistoryVerificationFullRunRechecksVerifiedEmpty(t *testing.T) { + t.Parallel() + + ctx, _, client, svc, channel := verificationFixture(t, []*discordgo.Message{storedMessage("300")}) + require.NoError(t, svc.store.SetSyncState(ctx, channelVerifiedEmptyScope("c1"), "1")) + + // A routine run trusts the marker. + count, err := svc.syncChannelMessages(ctx, "g1", channel, false, false, time.Time{}, false, nil) + require.NoError(t, err) + require.Zero(t, count) + require.Zero(t, client.messageCalls["c1"]) + + // An explicit full run re-checks it and recovers the message. + count, err = svc.syncChannelMessages(ctx, "g1", channel, true, false, time.Time{}, false, nil) + require.NoError(t, err) + require.Equal(t, 1, count) + require.Positive(t, client.messageCalls["c1"]) + + // The crawl disproved the marker, so it is retired in the same run. + marker, err := svc.store.GetSyncState(ctx, channelVerifiedEmptyScope("c1")) + require.NoError(t, err) + require.Empty(t, marker) +} + +func TestVerifiedEmptyNotWrittenForWindowedSync(t *testing.T) { + t.Parallel() + + // A windowed run must not verify at all: it can only reach back to the + // window, and completing there would lock the channel into a fraction of + // its history. It takes its unchanged pre-verification path instead. + ctx, s, client, svc, channel := verificationFixture(t, []*discordgo.Message{storedMessage("300")}) + since := time.Date(2027, 1, 1, 0, 0, 0, 0, time.UTC) + + count, err := svc.syncChannelMessages(ctx, "g1", channel, false, false, since, false, nil) + require.NoError(t, err) + require.Zero(t, count) + require.Zero(t, client.messageCalls["c1"], "a windowed run must not start a verification crawl") + + marker, err := s.GetSyncState(ctx, channelVerifiedEmptyScope("c1")) + require.NoError(t, err) + require.Empty(t, marker, "a since-windowed crawl must not mark a channel verified empty") + + // The untouched history_complete marker is the direct evidence that no + // verification began: verifyChannelHistory clears it before crawling. + complete, err := s.GetSyncState(ctx, channelHistoryCompleteScope("c1")) + require.NoError(t, err) + require.Equal(t, "1", complete) + + require.NoError(t, (*Syncer)(nil).recordVerifiedEmptyChannel(ctx, "c1")) + require.NoError(t, svc.recordVerifiedEmptyChannel(ctx, "")) + require.NoError(t, (*Syncer)(nil).restoreHistoryCompleteAfterFailedVerification(ctx, "c1")) + require.NoError(t, svc.restoreHistoryCompleteAfterFailedVerification(ctx, "")) +} + +func TestChannelHasMessagesProbe(t *testing.T) { + t.Parallel() + + ctx := context.Background() + s, err := store.Open(ctx, filepath.Join(t.TempDir(), "discrawl.db")) + require.NoError(t, err) + defer func() { _ = s.Close() }() + + has, err := s.ChannelHasMessages(ctx, "") + require.NoError(t, err) + require.False(t, has) + + has, err = s.ChannelHasMessages(ctx, "c1") + require.NoError(t, err) + require.False(t, has) + + require.NoError(t, s.UpsertChannel(ctx, store.ChannelRecord{ID: "c1", GuildID: "g1", Kind: "text", Name: "general", RawJSON: `{}`})) + require.NoError(t, s.UpsertMessage(ctx, store.MessageRecord{ + ID: "100", GuildID: "g1", ChannelID: "c1", ChannelName: "general", + AuthorID: "u1", AuthorName: "user", CreatedAt: time.Now().UTC().Format(time.RFC3339Nano), + Content: "hello", NormalizedContent: "hello", RawJSON: `{}`, + })) + + has, err = s.ChannelHasMessages(ctx, "c1") + require.NoError(t, err) + require.True(t, has) +} + +// LatestOnly is the default for routine syncs, so stranded channels reach the +// verification path through it. Verification ignores latest-only and crawls the +// whole channel; see TestHistoryVerificationLatestOnlyCrawlsWholeChannel for +// the multi-page case that separates the two. +func TestHistoryVerificationUnderLatestOnly(t *testing.T) { + t.Parallel() + + ctx, s, _, svc, channel := verificationFixture(t, []*discordgo.Message{storedMessage("300"), storedMessage("100")}) + + count, err := svc.syncChannelMessages(ctx, "g1", channel, false, false, time.Time{}, true, nil) + require.NoError(t, err) + require.Equal(t, 2, count) + + has, err := s.ChannelHasMessages(ctx, "c1") + require.NoError(t, err) + require.True(t, has) + + marker, err := s.GetSyncState(ctx, channelVerifiedEmptyScope("c1")) + require.NoError(t, err) + require.Empty(t, marker, "a channel that yielded messages must not be marked verified empty") +} + +func TestHistoryVerificationUnderLatestOnlyMarksEmptyChannel(t *testing.T) { + t.Parallel() + + ctx, s, client, svc, channel := verificationFixture(t, nil) + + count, err := svc.syncChannelMessages(ctx, "g1", channel, false, false, time.Time{}, true, nil) + require.NoError(t, err) + require.Zero(t, count) + firstRun := client.messageCalls["c1"] + require.Positive(t, firstRun) + + marker, err := s.GetSyncState(ctx, channelVerifiedEmptyScope("c1")) + require.NoError(t, err) + require.Equal(t, "1", marker) + + // And it does not loop on later runs. + _, err = svc.syncChannelMessages(ctx, "g1", channel, false, false, time.Time{}, true, nil) + require.NoError(t, err) + require.Equal(t, firstRun, client.messageCalls["c1"]) +} + +func TestHistoryVerificationUnderLatestOnlyForThread(t *testing.T) { + t.Parallel() + + // Threads already took the full-crawl branch before this change, and they + // still do: more than one page of history comes back whole. + ctx, s, _, svc, _ := verificationFixtureAt(t, storedMessages(250), "1249") + thread := &discordgo.Channel{ + ID: "c1", GuildID: "g1", ParentID: "f1", Name: "thread", + Type: discordgo.ChannelTypeGuildPublicThread, LastMessageID: "1249", + } + + count, err := svc.syncChannelMessages(ctx, "g1", thread, false, false, time.Time{}, true, nil) + require.NoError(t, err) + require.Equal(t, 250, count) + + oldest, newest, err := s.ChannelMessageBounds(ctx, "c1") + require.NoError(t, err) + require.Equal(t, "1000", oldest) + require.Equal(t, "1249", newest) +} + +// A non-thread channel in latest-only mode is the shape the one-page bug hit: +// the zeroed cursor sent it down syncLatestChannelHistory, which stored the +// newest page and stopped, and history_complete then made every later run skip +// it. A verification must crawl the channel to its first message. +func TestHistoryVerificationLatestOnlyCrawlsWholeChannel(t *testing.T) { + t.Parallel() + + ctx, s, client, svc, channel := verificationFixtureAt(t, storedMessages(250), "1249") + + count, err := svc.syncChannelMessages(ctx, "g1", channel, false, false, time.Time{}, true, nil) + require.NoError(t, err) + require.Equal(t, 250, count, "a verification must not stop after one page") + + // Reaching the oldest message is the proof the crawl ran to the start: the + // one-page path leaves the oldest stored id at the 151st message. + oldest, newest, err := s.ChannelMessageBounds(ctx, "c1") + require.NoError(t, err) + require.Equal(t, "1000", oldest) + require.Equal(t, "1249", newest) + + // The crawl reached the start, so it restored history_complete itself. + complete, err := s.GetSyncState(ctx, channelHistoryCompleteScope("c1")) + require.NoError(t, err) + require.Equal(t, "1", complete) + + marker, err := s.GetSyncState(ctx, channelVerifiedEmptyScope("c1")) + require.NoError(t, err) + require.Empty(t, marker) + + // The next run skips a channel that is now genuinely complete, and the + // history it locks in is the whole history. + calls := client.messageCalls["c1"] + count, err = svc.syncChannelMessages(ctx, "g1", channel, false, false, time.Time{}, true, nil) + require.NoError(t, err) + require.Zero(t, count) + require.Equal(t, calls, client.messageCalls["c1"]) + + oldest, _, err = s.ChannelMessageBounds(ctx, "c1") + require.NoError(t, err) + require.Equal(t, "1000", oldest) +} + +// A windowed run leaves a stranded channel exactly as it found it, including +// under latest-only, so nothing about it changes until an unwindowed run. +func TestHistoryVerificationSkippedForWindowedLatestOnlySync(t *testing.T) { + t.Parallel() + + ctx, s, client, svc, channel := verificationFixtureAt(t, storedMessages(250), "1249") + since := time.Date(2026, 5, 1, 0, 0, 0, 0, time.UTC) + + count, err := svc.syncChannelMessages(ctx, "g1", channel, false, false, since, true, nil) + require.NoError(t, err) + require.Zero(t, count) + require.Zero(t, client.messageCalls["c1"]) + + has, err := s.ChannelHasMessages(ctx, "c1") + require.NoError(t, err) + require.False(t, has) + + complete, err := s.GetSyncState(ctx, channelHistoryCompleteScope("c1")) + require.NoError(t, err) + require.Equal(t, "1", complete) + + marker, err := s.GetSyncState(ctx, channelVerifiedEmptyScope("c1")) + require.NoError(t, err) + require.Empty(t, marker) + + // The unwindowed run that follows still repairs it. + count, err = svc.syncChannelMessages(ctx, "g1", channel, false, false, time.Time{}, true, nil) + require.NoError(t, err) + require.Equal(t, 250, count) +} + +// A verification that fails before storing anything leaves the channel in the +// state that triggered it, so the next run tries again. +func TestHistoryVerificationRestoresMarkerWhenNothingWasStored(t *testing.T) { + t.Parallel() + + ctx, s, client, svc, channel := verificationFixtureAt(t, storedMessages(250), "1249") + client.messageErrors = map[string]error{"c1": errors.New("discord unavailable")} + + _, err := svc.syncChannelMessages(ctx, "g1", channel, false, false, time.Time{}, true, nil) + require.Error(t, err) + + complete, err := s.GetSyncState(ctx, channelHistoryCompleteScope("c1")) + require.NoError(t, err) + require.Equal(t, "1", complete, "a crawl that stored nothing must leave the channel verifiable") + + marker, err := s.GetSyncState(ctx, channelVerifiedEmptyScope("c1")) + require.NoError(t, err) + require.Empty(t, marker, "a failed crawl is not evidence of emptiness") + + delete(client.messageErrors, "c1") + count, err := svc.syncChannelMessages(ctx, "g1", channel, false, false, time.Time{}, true, nil) + require.NoError(t, err) + require.Equal(t, 250, count) +} + +// A verification that fails after storing part of the history must not leave +// history_complete over those partial rows: the rows make HasMessages true, so +// the marker would make every later run skip the channel for good. +func TestHistoryVerificationLeavesMarkerClearedAfterPartialCrawl(t *testing.T) { + t.Parallel() + + ctx, s, client, svc, channel := verificationFixtureAt(t, storedMessages(250), "1249") + client.beforeErrors = map[string]map[string]error{"c1": {"1150": errors.New("discord unavailable")}} + + _, err := svc.syncChannelMessages(ctx, "g1", channel, false, false, time.Time{}, true, nil) + require.Error(t, err) + + oldest, _, err := s.ChannelMessageBounds(ctx, "c1") + require.NoError(t, err) + require.Equal(t, "1150", oldest, "the first page should have been stored") + + complete, err := s.GetSyncState(ctx, channelHistoryCompleteScope("c1")) + require.NoError(t, err) + require.Empty(t, complete, "a partial history must not be marked complete") + + cursor, err := s.GetSyncState(ctx, channelBackfillScope("c1")) + require.NoError(t, err) + require.Equal(t, "1150", cursor) + + // The channel is a resumable backfill, so a full run finishes it. + delete(client.beforeErrors, "c1") + count, err := svc.syncChannelMessages(ctx, "g1", channel, true, false, time.Time{}, false, nil) + require.NoError(t, err) + require.Equal(t, 150, count) + + oldest, _, err = s.ChannelMessageBounds(ctx, "c1") + require.NoError(t, err) + require.Equal(t, "1000", oldest) + + complete, err = s.GetSyncState(ctx, channelHistoryCompleteScope("c1")) + require.NoError(t, err) + require.Equal(t, "1", complete) +} + +// The per-channel deadline is the likeliest way a full verification crawl +// fails, so the marker restore must survive a cancelled context. +func TestHistoryVerificationRestoresMarkerAfterContextDeadline(t *testing.T) { + t.Parallel() + + ctx, s, client, svc, channel := verificationFixtureAt(t, storedMessages(250), "1249") + blocked := make(chan struct{}) + defer close(blocked) + client.messageBlocks = map[string]chan struct{}{"c1": blocked} + + crawlCtx, cancel := context.WithTimeout(ctx, 100*time.Millisecond) + defer cancel() + _, err := svc.syncChannelMessages(crawlCtx, "g1", channel, false, false, time.Time{}, true, nil) + require.ErrorIs(t, err, context.DeadlineExceeded) + + complete, err := s.GetSyncState(ctx, channelHistoryCompleteScope("c1")) + require.NoError(t, err) + require.Equal(t, "1", complete, "a crawl killed by its deadline must leave the channel verifiable") +} + +// verified_empty must describe the channel as it is now, not as it was. A +// channel that gains messages after being marked keeps no marker, so losing +// those messages later still triggers a verification. +func TestVerifiedEmptyMarkerClearedOnceMessagesArrive(t *testing.T) { + t.Parallel() + + ctx, s, client, svc, channel := verificationFixture(t, []*discordgo.Message{storedMessage("400")}) + require.NoError(t, s.SetSyncState(ctx, channelVerifiedEmptyScope("c1"), "1")) + channel.LastMessageID = "400" + + // An ordinary incremental run stores the new message. It never consults + // verified_empty and never calls recordVerifiedEmptyChannel, which is why + // clearing the marker there alone would leave this channel stale. + count, err := svc.syncChannelMessages(ctx, "g1", channel, false, false, time.Time{}, true, nil) + require.NoError(t, err) + require.Equal(t, 1, count) + has, err := s.ChannelHasMessages(ctx, "c1") + require.NoError(t, err) + require.True(t, has) + + // Whatever that run left behind, the next load reconciles the marker + // against the rows now on disk. + _, err = svc.syncChannelMessages(ctx, "g1", channel, false, false, time.Time{}, true, nil) + require.NoError(t, err) + marker, err := s.GetSyncState(ctx, channelVerifiedEmptyScope("c1")) + require.NoError(t, err) + require.Empty(t, marker, "a channel holding messages must not stay marked verified empty") + + // Losing the rows again puts the channel back into the stranded shape, and + // with no stale marker in the way it is verified rather than skipped. + require.NoError(t, s.DeleteGuildData(ctx, "g1")) + require.NoError(t, s.UpsertChannel(ctx, store.ChannelRecord{ID: "c1", GuildID: "g1", Kind: "text", Name: "general", RawJSON: `{}`})) + calls := client.messageCalls["c1"] + + count, err = svc.syncChannelMessages(ctx, "g1", channel, false, false, time.Time{}, true, nil) + require.NoError(t, err) + require.Equal(t, 1, count) + require.Greater(t, client.messageCalls["c1"], calls, "the channel must be re-crawled, not skipped") + has, err = s.ChannelHasMessages(ctx, "c1") + require.NoError(t, err) + require.True(t, has) +} diff --git a/internal/syncer/message_sync.go b/internal/syncer/message_sync.go index 84e7784f..0e3c1045 100644 --- a/internal/syncer/message_sync.go +++ b/internal/syncer/message_sync.go @@ -2,6 +2,7 @@ package syncer import ( "context" + "errors" "fmt" "sync" "time" @@ -224,16 +225,123 @@ func (s *Syncer) syncChannelMessages(ctx context.Context, guildID string, channe if err != nil { return 0, err } + if state.HasMessages && state.VerifiedEmpty { + // The channel holds rows, so the marker describes a state that has + // since gone away. Clearing it here, before anything is decided from + // it, is what keeps it honest on every path: rows also arrive from the + // gateway tail and from ordinary incremental syncs, neither of which + // passes through verification, and a marker left over from before them + // would suppress the verification that recovers those rows if they are + // lost again. + if err := s.store.DeleteSyncState(ctx, channelVerifiedEmptyScope(channel.ID)); err != nil { + return 0, err + } + state.VerifiedEmpty = false + } + if full { + // An explicit full run re-checks a channel previously verified empty. + state.VerifiedEmpty = false + } + if needsHistoryVerification(channel, state, since) { + return s.verifyChannelHistory(ctx, channel, embeddings, progress) + } + return s.syncChannelHistory(ctx, channel, state, full, embeddings, since, latestOnly, progress) +} + +// verifyChannelHistory re-crawls a channel that carries history_complete while +// holding no message rows, and records the outcome. It always crawls the whole +// channel: routing a verification through the latest-only or incremental paths +// stores a single page (or a since window) and then lets history_complete lock +// that partial history in, because the next run sees stored rows and skips the +// channel for good. Forcing the full path here matches what syncChannelHistory +// already does for an incomplete thread. +func (s *Syncer) verifyChannelHistory(ctx context.Context, channel *discordgo.Channel, embeddings bool, progress *messageSyncProgress) (int, error) { + // The stored cursors describe a history that is not actually present, so + // crawl from scratch with an empty state. Dropping the in-memory copy is + // safe: AdvanceChannelLatestMessageID only ever moves the stored pointer + // forward, so it cannot be rewound from here. + // + // history_complete is cleared for the duration of the crawl. Leaving it in + // place strands the channel if the crawl fails partway: the rows it did + // store make HasMessages true, so the next run neither verifies nor skips + // its way back into a backfill. A crawl that reaches the start of the + // channel sets the marker again itself (syncBackfillPages), and + // verification only ever runs unwindowed, so success always restores it. A + // process killed mid-crawl leaves the marker off, which is the same + // resumable state as any other interrupted backfill. + if err := s.store.DeleteSyncState(ctx, channelHistoryCompleteScope(channel.ID)); err != nil { + return 0, err + } + count, err := s.syncFullChannelHistory(ctx, channel, channelSyncState{}, embeddings, time.Time{}, progress) + if err != nil { + // The crawl now spans every page rather than one, so the per-channel + // deadline in messageChannelContext is a likely way for it to fail. + // The restore has to outlive that: on a cancelled context both of its + // queries fail, and the marker this function removed stays removed. + return count, errors.Join(err, s.restoreHistoryCompleteAfterFailedVerification(context.WithoutCancel(ctx), channel.ID)) + } + return count, s.recordVerifiedEmptyChannel(ctx, channel.ID) +} + +// restoreHistoryCompleteAfterFailedVerification puts history_complete back when +// a verification crawl failed without storing anything. The channel is then in +// exactly the state that triggered verification, so the next run looks at it +// again instead of treating an untouched channel as incomplete. When the failed +// crawl did store rows the marker stays cleared: re-marking a partial history +// complete is the lock-in this path exists to prevent, and the channel is left +// as a resumable backfill that a full run continues from its cursor. +func (s *Syncer) restoreHistoryCompleteAfterFailedVerification(ctx context.Context, channelID string) error { + if s == nil || s.store == nil || channelID == "" { + return nil + } + // The store is the authority here, not the returned message count: a page + // can be committed by persistMessagePage and still report zero if a later + // step in the same call fails. + hasMessages, err := s.store.ChannelHasMessages(ctx, channelID) + if err != nil { + return err + } + if hasMessages { + return nil + } + return s.store.SetSyncState(ctx, channelHistoryCompleteScope(channelID), "1") +} + +// recordVerifiedEmptyChannel records the outcome of a verification crawl. A +// crawl that stored nothing marks the channel, so needsHistoryVerification +// stops re-crawling it on every run; a crawl that recovered rows clears any +// marker instead, because a channel that holds messages is not empty and a +// marker saying otherwise would suppress a later recovery. It needs no since +// guard of its own: needsHistoryVerification +// declines to verify a windowed sync at all, so a channel can never be recorded +// empty because filterMessagesSince dropped everything before it was persisted. +func (s *Syncer) recordVerifiedEmptyChannel(ctx context.Context, channelID string) error { + if s == nil || s.store == nil || channelID == "" { + return nil + } + hasMessages, err := s.store.ChannelHasMessages(ctx, channelID) + if err != nil { + return err + } + if hasMessages { + // The crawl disproved the marker, so retire it in the same run rather + // than leaving a stale one for the next load to reconcile. + return s.store.DeleteSyncState(ctx, channelVerifiedEmptyScope(channelID)) + } + return s.store.SetSyncState(ctx, channelVerifiedEmptyScope(channelID), "1") +} + +func (s *Syncer) syncChannelHistory(ctx context.Context, channel *discordgo.Channel, state channelSyncState, full bool, embeddings bool, since time.Time, latestOnly bool, progress *messageSyncProgress) (int, error) { if full { if err := s.seedChannelSyncState(ctx, channel.ID, &state); err != nil { return 0, err } - if shouldSkipChannelSync(channel, state) { + if shouldSkipChannelSync(channel, state, since) { return 0, nil } return s.syncFullChannelHistory(ctx, channel, state, embeddings, since, progress) } - if shouldSkipChannelSync(channel, state) { + if shouldSkipChannelSync(channel, state, since) { return 0, nil } if latestOnly { @@ -255,12 +363,48 @@ type channelSyncState struct { StoredLatest string BackfillCursor string BackfillComplete bool + // HasMessages reports whether any message rows exist locally for the + // channel. history_complete alone is not evidence that the history was + // actually stored: a channel can carry the marker with zero local rows. + HasMessages bool + // VerifiedEmpty reports that a previous verification pass re-fetched the + // channel from scratch and still found nothing, so it must not be + // re-fetched every run. + VerifiedEmpty bool } -func shouldSkipChannelSync(channel *discordgo.Channel, state channelSyncState) bool { +// needsHistoryVerification reports whether a channel marked history_complete +// must be re-fetched because nothing is stored locally while Discord still +// reports the channel holds content. Without this, a channel whose messages are +// missing is skipped forever, including under --full. +func needsHistoryVerification(channel *discordgo.Channel, state channelSyncState, since time.Time) bool { + if channel == nil || !state.BackfillComplete { + return false + } + if state.HasMessages || state.VerifiedEmpty { + return false + } + // A windowed run cannot complete a recovery: it can only fetch back to the + // window, and finishing there would mark a fraction of the history + // complete. Leave the channel on its normal path instead. Routine syncs + // carry no --since, so a stranded channel is still repaired by the next + // unwindowed run, and declining here costs nothing beyond that delay while + // keeping a deliberately narrow sync from turning into a full crawl. + if !since.IsZero() { + return false + } + // Discord reporting no last message is consistent with an empty channel, + // so there is nothing to recover. + return channel.LastMessageID != "" +} + +func shouldSkipChannelSync(channel *discordgo.Channel, state channelSyncState, since time.Time) bool { if !state.BackfillComplete || channel == nil { return false } + if needsHistoryVerification(channel, state, since) { + return false + } if channel.LastMessageID == "" { return state.Latest == "" } @@ -290,12 +434,32 @@ func (s *Syncer) loadChannelSyncState(ctx context.Context, channelID string) (ch if err != nil { return channelSyncState{}, err } - return channelSyncState{ + state := channelSyncState{ Latest: latest, StoredLatest: latest, BackfillCursor: backfillCursor, BackfillComplete: backfillComplete != "", - }, nil + } + if !state.BackfillComplete { + // Only a completed channel can be wrongly trusted, so the probes below + // are unnecessary work for every other channel. + return state, nil + } + hasMessages, err := s.store.ChannelHasMessages(ctx, channelID) + if err != nil { + return channelSyncState{}, err + } + state.HasMessages = hasMessages + // The marker is read even when rows are stored, so a stale one can be + // spotted and cleared. It is a point read on the sync_state primary key, + // paid only by channels already marked complete, and it is the single place + // the marker is read, so no caller can act on one this load did not see. + verifiedEmpty, err := s.store.GetSyncState(ctx, channelVerifiedEmptyScope(channelID)) + if err != nil { + return channelSyncState{}, err + } + state.VerifiedEmpty = verifiedEmpty != "" + return state, nil } func (s *Syncer) seedChannelSyncState(ctx context.Context, channelID string, state *channelSyncState) error { diff --git a/internal/syncer/message_sync_helpers_test.go b/internal/syncer/message_sync_helpers_test.go index a02f6f5b..c8a20b38 100644 --- a/internal/syncer/message_sync_helpers_test.go +++ b/internal/syncer/message_sync_helpers_test.go @@ -54,10 +54,10 @@ func TestChannelSyncStateHelpers(t *testing.T) { t.Parallel() channel := &discordgo.Channel{ID: "c1", LastMessageID: "200"} - require.False(t, shouldSkipChannelSync(nil, channelSyncState{BackfillComplete: true})) - require.True(t, shouldSkipChannelSync(&discordgo.Channel{ID: "c1"}, channelSyncState{BackfillComplete: true, Latest: ""})) - require.False(t, shouldSkipChannelSync(channel, channelSyncState{BackfillComplete: true, Latest: ""})) - require.True(t, shouldSkipChannelSync(channel, channelSyncState{BackfillComplete: true, Latest: "300"})) + require.False(t, shouldSkipChannelSync(nil, channelSyncState{BackfillComplete: true}, time.Time{})) + require.True(t, shouldSkipChannelSync(&discordgo.Channel{ID: "c1"}, channelSyncState{BackfillComplete: true, Latest: ""}, time.Time{})) + require.False(t, shouldSkipChannelSync(channel, channelSyncState{BackfillComplete: true, Latest: ""}, time.Time{})) + require.True(t, shouldSkipChannelSync(channel, channelSyncState{BackfillComplete: true, Latest: "300", HasMessages: true}, time.Time{})) require.False(t, shouldSkipLatestOnlyChannelSync(nil, channelSyncState{Latest: "300"})) require.False(t, shouldSkipLatestOnlyChannelSync(channel, channelSyncState{})) require.True(t, shouldSkipLatestOnlyChannelSync(channel, channelSyncState{Latest: "300"})) @@ -111,7 +111,7 @@ func TestChannelSyncStateStoreHelpers(t *testing.T) { require.NoError(t, s.SetSyncState(ctx, channelHistoryCompleteScope("c1"), "1")) loaded, err := svc.loadChannelSyncState(ctx, "c1") require.NoError(t, err) - require.Equal(t, channelSyncState{Latest: "200", StoredLatest: "200", BackfillCursor: "100", BackfillComplete: true}, loaded) + require.Equal(t, channelSyncState{Latest: "200", StoredLatest: "200", BackfillCursor: "100", BackfillComplete: true, HasMessages: true}, loaded) } func TestMessageChannelSyncBranches(t *testing.T) { diff --git a/internal/syncer/records.go b/internal/syncer/records.go index 50182ed8..2a83d42f 100644 --- a/internal/syncer/records.go +++ b/internal/syncer/records.go @@ -212,6 +212,14 @@ func channelHistoryCompleteScope(channelID string) string { return "channel:" + channelID + ":history_complete" } +// channelVerifiedEmptyScope records that a channel marked history_complete was +// re-fetched from scratch and genuinely yielded no messages. It stops the +// history verification in needsHistoryVerification from re-fetching such a +// channel on every run. +func channelVerifiedEmptyScope(channelID string) string { + return "channel:" + channelID + ":verified_empty" +} + func channelMessageUnavailableScope(channelID string) string { return "channel:" + channelID + ":unavailable" }