diff --git a/CHANGELOG.md b/CHANGELOG.md index d6aefde..e822945 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ ### Fixes +- Prevent global coverage reports from timing out on large archives by aggregating messages in one sequential scan and allowing a longer bounded deadline. - Preserve guild and member history with source-attributed tombstones, explicit restore semantics, omission-safe member refreshes, and revision-aware Git-share merges. - Keep attachment fetches compatible with three allowed Discord CDN redirects and injected HTTP transports while preserving final-response host validation. - Reject malformed tombstone timestamps before routine incremental snapshot imports mutate the archive. diff --git a/internal/store/coverage.go b/internal/store/coverage.go index 62b0431..e23171a 100644 --- a/internal/store/coverage.go +++ b/internal/store/coverage.go @@ -11,7 +11,50 @@ import ( const wiretapStatsScope = "wiretap:last_stats:v1" -const coverageQueryTimeout = 2 * time.Minute +const ( + coverageQueryTimeout = 2 * time.Minute + globalCoverageQueryTimeout = 8 * time.Minute +) + +const filteredCoverageChannelQuery = ` + select + c.id, c.guild_id, c.name, c.kind, + case when exists ( + select 1 from sync_state s + where s.scope = 'channel:' || c.id || ':history_complete' + ) then 1 else 0 end, + count(m.id), coalesce(min(m.created_at), ''), coalesce(max(m.created_at), '') + from channels c + left join messages m on m.channel_id = c.id and m.deleted_at is null + where c.guild_id = ? + group by c.id, c.guild_id, c.name, c.kind + order by c.guild_id, count(m.id) desc, lower(c.name), c.id +` + +const globalCoverageChannelQuery = ` + with message_coverage as materialized ( + select + channel_id, + count(*) as message_count, + coalesce(min(created_at), '') as earliest_message_at, + coalesce(max(created_at), '') as latest_message_at + from messages not indexed + where deleted_at is null + group by channel_id + ) + select + c.id, c.guild_id, c.name, c.kind, + case when exists ( + select 1 from sync_state s + where s.scope = 'channel:' || c.id || ':history_complete' + ) then 1 else 0 end, + coalesce(m.message_count, 0), + coalesce(m.earliest_message_at, ''), + coalesce(m.latest_message_at, '') + from channels c + left join message_coverage m on m.channel_id = c.id + order by c.guild_id, coalesce(m.message_count, 0) desc, lower(c.name), c.id +` var messageChannelKinds = map[string]struct{}{ "text": {}, "news": {}, "announcement": {}, "dm": {}, "group_dm": {}, @@ -102,7 +145,7 @@ func (s *Store) SetWiretapImportStats(ctx context.Context, stats WiretapImportSt func (s *Store) Coverage(ctx context.Context, guildID string, generatedAt time.Time) (CoverageReport, error) { report := CoverageReport{GeneratedAt: generatedAt.UTC(), Guilds: []CoverageGuild{}} - queryCtx, cancel := context.WithTimeout(ctx, coverageQueryTimeout) + queryCtx, cancel := withCoverageQueryTimeout(ctx, guildID) defer cancel() rows, err := s.db.QueryContext(queryCtx, ` @@ -137,20 +180,13 @@ func (s *Store) Coverage(ctx context.Context, guildID string, generatedAt time.T for i := range report.Guilds { guilds[report.Guilds[i].ID] = &report.Guilds[i] } - channelRows, err := s.db.QueryContext(queryCtx, ` - select - c.id, c.guild_id, c.name, c.kind, - case when exists ( - select 1 from sync_state s - where s.scope = 'channel:' || c.id || ':history_complete' - ) then 1 else 0 end, - count(m.id), coalesce(min(m.created_at), ''), coalesce(max(m.created_at), '') - from channels c - left join messages m on m.channel_id = c.id and m.deleted_at is null - where ? = '' or c.guild_id = ? - group by c.id, c.guild_id, c.name, c.kind - order by c.guild_id, count(m.id) desc, lower(c.name), c.id - `, guildID, guildID) + channelQuery := filteredCoverageChannelQuery + channelArgs := []any{guildID} + if guildID == "" { + channelQuery = globalCoverageChannelQuery + channelArgs = nil + } + channelRows, err := s.db.QueryContext(queryCtx, channelQuery, channelArgs...) if err != nil { return CoverageReport{}, fmt.Errorf("query channel coverage: %w", err) } @@ -226,6 +262,14 @@ func (s *Store) Coverage(ctx context.Context, guildID string, generatedAt time.T return report, nil } +func withCoverageQueryTimeout(ctx context.Context, guildID string) (context.Context, context.CancelFunc) { + timeout := coverageQueryTimeout + if guildID == "" { + timeout = globalCoverageQueryTimeout + } + return context.WithTimeout(ctx, timeout) +} + func (s *Store) loadKnownFailureCoverage(ctx context.Context, guildID string, report *CoverageReport) error { rows, err := s.db.QueryContext(ctx, ` select guild_id, channel_id, count(*) diff --git a/internal/store/coverage_test.go b/internal/store/coverage_test.go index 8af199c..0835225 100644 --- a/internal/store/coverage_test.go +++ b/internal/store/coverage_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "path/filepath" + "strings" "testing" "time" @@ -21,14 +22,19 @@ func TestCoverageReportsGuildChannelAndWiretapState(t *testing.T) { require.NoError(t, s.UpsertGuild(ctx, GuildRecord{ID: "g2", Name: "Guild Two", RawJSON: `{}`})) require.NoError(t, s.UpsertChannel(ctx, ChannelRecord{ID: "c1", GuildID: "g1", Kind: "text", Name: "general", RawJSON: `{}`})) require.NoError(t, s.UpsertChannel(ctx, ChannelRecord{ID: "c2", GuildID: "g1", Kind: "text", Name: "channel-c2", RawJSON: `{"source":"discord_desktop"}`})) + require.NoError(t, s.UpsertChannel(ctx, ChannelRecord{ID: "c3", GuildID: "g2", Kind: "text", Name: "empty", RawJSON: `{}`})) require.NoError(t, s.UpsertChannel(ctx, ChannelRecord{ID: "v1", GuildID: "g1", Kind: "voice", Name: "Voice", RawJSON: `{}`})) for _, message := range []MessageRecord{ + {ID: "deleted-early", GuildID: "g1", ChannelID: "c1", CreatedAt: "2026-05-01T10:00:00Z", Content: "deleted early", NormalizedContent: "deleted early", RawJSON: `{}`}, {ID: "m1", GuildID: "g1", ChannelID: "c1", CreatedAt: "2026-06-01T10:00:00Z", Content: "one", NormalizedContent: "one", RawJSON: `{}`}, {ID: "m2", GuildID: "g1", ChannelID: "c1", CreatedAt: "2026-06-02T10:00:00Z", Content: "two", NormalizedContent: "two", RawJSON: `{}`}, {ID: "m3", GuildID: "g1", ChannelID: "c2", CreatedAt: "2026-06-03T10:00:00Z", Content: "three", NormalizedContent: "three", RawJSON: `{}`}, + {ID: "deleted-late", GuildID: "g1", ChannelID: "c1", CreatedAt: "2026-07-01T10:00:00Z", Content: "deleted late", NormalizedContent: "deleted late", RawJSON: `{}`}, } { require.NoError(t, s.UpsertMessage(ctx, message)) } + require.NoError(t, s.MarkMessageDeleted(ctx, "g1", "c1", "deleted-early", nil)) + require.NoError(t, s.MarkMessageDeleted(ctx, "g1", "c1", "deleted-late", nil)) require.NoError(t, s.SetSyncState(ctx, "channel:c1:history_complete", "1")) require.NoError(t, s.SetSyncState(ctx, "sync:last_success", "2026-06-04T10:00:00Z")) require.NoError(t, s.SetSyncState(ctx, "wiretap:last_import", "2026-06-04T11:00:00Z")) @@ -44,8 +50,8 @@ func TestCoverageReportsGuildChannelAndWiretapState(t *testing.T) { require.NoError(t, err) require.Equal(t, generatedAt, report.GeneratedAt) require.Equal(t, CoverageTotals{ - GuildCount: 2, MessageCount: 3, ChannelCount: 3, MessageChannelCount: 2, - NamedChannelCount: 2, SyntheticChannelCount: 1, HistoryCompleteChannelCount: 1, + GuildCount: 2, MessageCount: 3, ChannelCount: 4, MessageChannelCount: 3, + NamedChannelCount: 3, SyntheticChannelCount: 1, HistoryCompleteChannelCount: 1, KnownFailureCount: 2, UnscopedKnownFailureCount: 1, }, report.Totals) require.Equal(t, time.Date(2026, 6, 4, 10, 0, 0, 0, time.UTC), report.LastBotSyncAt) @@ -63,15 +69,92 @@ func TestCoverageReportsGuildChannelAndWiretapState(t *testing.T) { require.Equal(t, 1, report.Guilds[0].KnownFailureCount) require.Equal(t, 1, report.Guilds[0].Channels[0].KnownFailureCount) require.True(t, report.Guilds[0].Channels[1].Synthetic) + require.Equal(t, "v1", report.Guilds[0].Channels[2].ID) + require.Equal(t, 0, report.Guilds[0].Channels[2].MessageCount) - filtered, err := s.Coverage(ctx, "g2", generatedAt) + filtered, err := s.Coverage(ctx, "g1", generatedAt) require.NoError(t, err) - require.Equal(t, CoverageTotals{GuildCount: 1}, filtered.Totals) - require.Equal(t, "g2", filtered.Guilds[0].ID) + require.Equal(t, 3, filtered.Totals.MessageCount) + require.Equal(t, 3, filtered.Totals.ChannelCount) + require.Equal(t, "g1", filtered.Guilds[0].ID) + require.Equal(t, report.Guilds[0].Channels, filtered.Guilds[0].Channels) + + empty, err := s.Coverage(ctx, "g2", generatedAt) + require.NoError(t, err) + require.Equal(t, CoverageTotals{ + GuildCount: 1, ChannelCount: 1, MessageChannelCount: 1, NamedChannelCount: 1, + }, empty.Totals) + require.Equal(t, "c3", empty.Guilds[0].Channels[0].ID) + require.Equal(t, 0, empty.Guilds[0].Channels[0].MessageCount) _, err = s.Coverage(ctx, "missing", generatedAt) require.ErrorContains(t, err, `guild "missing" not found`) } +func TestCoverageQueryTimeoutsAndCancellation(t *testing.T) { + t.Parallel() + + started := time.Now() + globalCtx, globalCancel := withCoverageQueryTimeout(context.Background(), "") + globalDeadline, ok := globalCtx.Deadline() + require.True(t, ok) + require.WithinDuration(t, started.Add(8*time.Minute), globalDeadline, time.Second) + globalCancel() + + started = time.Now() + filteredCtx, filteredCancel := withCoverageQueryTimeout(context.Background(), "g1") + filteredDeadline, ok := filteredCtx.Deadline() + require.True(t, ok) + require.WithinDuration(t, started.Add(2*time.Minute), filteredDeadline, time.Second) + filteredCancel() + + parentDeadline := time.Now().Add(time.Minute) + parentCtx, parentCancel := context.WithDeadline(context.Background(), parentDeadline) + defer parentCancel() + childCtx, childCancel := withCoverageQueryTimeout(parentCtx, "") + defer childCancel() + childDeadline, ok := childCtx.Deadline() + require.True(t, ok) + require.Equal(t, parentDeadline, childDeadline) + + ctx := context.Background() + s, err := Open(ctx, filepath.Join(t.TempDir(), "discrawl.db")) + require.NoError(t, err) + defer func() { _ = s.Close() }() + + for _, guildID := range []string{"", "g1"} { + canceledCtx, cancel := context.WithCancel(ctx) + cancel() + _, err := s.Coverage(canceledCtx, guildID, time.Now()) + require.ErrorIs(t, err, context.Canceled) + } +} + +func TestCoverageGlobalQueryPlanMaterializesSequentialMessageScan(t *testing.T) { + ctx := context.Background() + s, err := Open(ctx, filepath.Join(t.TempDir(), "discrawl.db")) + require.NoError(t, err) + defer func() { require.NoError(t, s.Close()) }() + + rows, err := s.DB().QueryContext(ctx, "explain query plan "+globalCoverageChannelQuery) + require.NoError(t, err) + defer func() { require.NoError(t, rows.Close()) }() + + var details []string + for rows.Next() { + var selectID, parentID, unused int + var detail string + require.NoError(t, rows.Scan(&selectID, &parentID, &unused, &detail)) + details = append(details, detail) + } + require.NoError(t, rows.Err()) + + plan := strings.ToLower(strings.Join(details, "\n")) + require.Contains(t, plan, "materialize message_coverage") + require.Contains(t, plan, "scan messages") + require.NotContains(t, plan, "idx_messages_channel_id") + require.NotContains(t, plan, "idx_messages_channel_created_id") +} + func TestCoverageDeltaSince(t *testing.T) { previous := CoverageReport{Totals: CoverageTotals{MessageCount: 4, ChannelCount: 3, NamedChannelCount: 2, SyntheticChannelCount: 1}} current := CoverageReport{Totals: CoverageTotals{MessageCount: 7, ChannelCount: 4, NamedChannelCount: 4, SyntheticChannelCount: 0}}