feat: recover restart gaps when tail connects

CHANGELOG.md

@@ -6,6 +6,8 @@
 - Use CrawlKit v0.16.1 to bound live-embedding shutdown when cancellation interrupts retry or completion persistence and cleanup storage is unavailable.
 - Generate deterministic, aggregate-only Discord field notes alongside daily published activity reports, with paired Markdown/JSON artifacts, separate markers preserving legacy narrative notes, explicit timestamp and coverage limits, and filtered-publication cleanup.
 
+- Add opt-in `tail --repair-on-start` to recover restart gaps immediately after Gateway connection, using the existing serialized repair and writer ownership.
+
 ## 0.14.1 - 2026-09-09
 
 - Accept the application's legacy migrated attachment column order in the backup-first repair helper while retaining exact schema and repair safeguards.

docs/commands/tail.md

@@ -11,6 +11,7 @@ discrawl tail --with-embeddings
 discrawl tail --embed-live
 discrawl tail --guild 123456789012345678
 discrawl tail --repair-every 30m
+discrawl tail --repair-on-start --repair-every 6h
 discrawl tail --replay-failures-only
 ```
 
@@ -28,6 +29,7 @@ discrawl tail --replay-failures-only
 
 - `--guild <id>` / `--guilds <id,id>` - tail a specific guild scope (default: `default_guild_id`, or all discovered guilds if unset)
 - `--repair-every <duration>` - frequency of the repair sweep
+- `--repair-on-start` - run one catch-up repair after the Gateway connects, without waiting for the periodic timer (default: off; also works with `--repair-every 0`)
 - `--embed-live` - continuously process queued embeddings while capture continues (opt-in; implies queueing; requires configured embeddings)
 - `--with-embeddings` - queue live, replayed, and repair messages for embedding (default: off)
 - `--replay-failures-only` - replay unresolved exact-message tail failures and exit
@@ -36,6 +38,8 @@ discrawl tail --replay-failures-only
 ## Notes
 
 - requires a working Discord bot token
+- startup repair uses the same writer owner and serialized repair lifecycle as periodic repair; capture remains connected while missed history is fetched, and shutdown cancels and joins the repair
+- with `--repair-on-start`, REST repair owns history cursors for that tail session; live events still update messages and live freshness immediately, but cannot advance history progress past missing messages. Periodic and subsequent startup repairs may therefore re-fetch already captured messages. This also preserves catch-up progress if repair is interrupted.
 - not available in Git-only mode (`discord.token_source = "none"`)
 - `discrawl --verbose tail` traces Gateway receipt, worker handling, scope
   filtering, and successful archive writes with event and Discord IDs but no

internal/cli/admin_commands.go

@@ -329,6 +329,7 @@ func (r *runtime) runTail(args []string) error {
 	fs := flag.NewFlagSet("tail", flag.ContinueOnError)
 	fs.SetOutput(io.Discard)
 	repairEvery := fs.Duration("repair-every", mustDuration(r.cfg.Sync.RepairEvery), "")
+	repairOnStart := fs.Bool("repair-on-start", false, "")
 	withEmbeddings := fs.Bool("with-embeddings", false, "")
 	embedLive := fs.Bool("embed-live", false, "")
 	replayFailuresOnly := fs.Bool("replay-failures-only", false, "")
@@ -357,6 +358,14 @@ func (r *runtime) runTail(args []string) error {
 	if *embedLive && *replayFailuresOnly {
 		return usageErr(errors.New("--embed-live cannot be combined with --replay-failures-only"))
 	}
+	if *repairOnStart && *replayFailuresOnly {
+		return usageErr(errors.New("--repair-on-start cannot be combined with --replay-failures-only"))
+	}
+	if configurable, ok := r.syncer.(tailStartupRepairConfigurer); ok {
+		configurable.SetTailRepairOnStart(*repairOnStart)
+	} else if *repairOnStart {
+		return errors.New("startup tail repair is unavailable")
+	}
 	if *embedLive && !r.cfg.Search.Embeddings.Enabled {
 		return usageErr(errors.New("--embed-live requires embeddings enabled in config"))
 	}

internal/cli/cli.go

@@ -306,6 +306,10 @@ type tailEmbeddingsConfigurer interface {
 	SetTailEmbeddings(bool)
 }
 
+type tailStartupRepairConfigurer interface {
+	SetTailRepairOnStart(bool)
+}
+
 type tailMessageFailureReplayer interface {
 	ReplayTailMessageFailures(context.Context, []string, int) (syncer.TailMessageReplayStats, error)
 }

internal/cli/cli_test.go

@@ -3692,6 +3692,7 @@ type fakeSyncService struct {
 	includeCategoryIDs    []string
 	repairOffset          time.Duration
 	tailEmbeddings        bool
+	tailRepairOnStart     bool
 	callTailReady         bool
 	tailReadyCalls        int
 	tailReady             func(context.Context) error
@@ -3742,6 +3743,10 @@ func (f *fakeSyncService) SetTailEmbeddings(enabled bool) {
 	f.tailEmbeddings = enabled
 }
 
+func (f *fakeSyncService) SetTailRepairOnStart(enabled bool) {
+	f.tailRepairOnStart = enabled
+}
+
 func (f *fakeSyncService) SetAttachmentTextEnabled(enabled bool) {
 	f.attachmentTextEnabled = enabled
 }

internal/cli/output.go

@@ -130,11 +130,12 @@ Sync Discord or desktop-cache data into the local archive.
 Configure Discord collection scope with sync.include_category_ids,
 sync.exclude_channel_ids, and sync.exclude_channel_kinds; exclusions win.
 `,
-	"tail": `Usage: discrawl tail [--repair-every DURATION] [--with-embeddings] [--embed-live] [--guild ID|--guilds IDS] [--replay-failures-only [--replay-limit N]]
+	"tail": `Usage: discrawl tail [--repair-every DURATION] [--repair-on-start] [--with-embeddings] [--embed-live] [--guild ID|--guilds IDS] [--replay-failures-only [--replay-limit N]]
 
 Continuously archive new Discord messages.
 Use --with-embeddings to queue live, replayed, and repair messages for embedding.
 Use --embed-live to also process embeddings continuously without pausing capture.
+Use --repair-on-start to catch up missed history as soon as the Gateway is connected.
 The sync.include_category_ids, sync.exclude_channel_ids, and
 sync.exclude_channel_kinds settings apply to live events and repair syncs.
 `,

internal/cli/startup_repair_test.go

@@ -0,0 +1,24 @@
+package cli
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestTailStartupRepairIsExplicitAndIncompatibleWithReplayOnly(t *testing.T) {
+	for _, args := range [][]string{{"tail"}, {"tail", "--repair-on-start", "--repair-every", "0"}, {"tail", "--repair-on-start", "--replay-failures-only"}} {
+		_, path := writeTestConfig(t, t.TempDir())
+		fake := &fakeSyncService{callTailReady: true}
+		rt := tailTestRuntime(t.Context(), path, fake)
+		err := rt.dispatch(args)
+		if args[len(args)-1] == "--replay-failures-only" {
+			require.ErrorContains(t, err, "cannot be combined")
+			require.Zero(t, fake.tailCalls)
+			continue
+		}
+		require.NoError(t, err)
+		require.Equal(t, len(args) > 1, fake.tailRepairOnStart)
+		require.Equal(t, 1, fake.tailCalls)
+	}
+}

internal/syncer/startup_repair_test.go

@@ -0,0 +1,217 @@
+package syncer
+
+import (
+	"context"
+	"fmt"
+	"path/filepath"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/bwmarrin/discordgo"
+	discordclient "github.com/openclaw/discrawl/internal/discord"
+	"github.com/openclaw/discrawl/internal/store"
+	"github.com/stretchr/testify/require"
+)
+
+type startupTailClient struct {
+	*fakeClient
+	serve func(context.Context, discordclient.EventHandler) error
+}
+
+func (c *startupTailClient) Tail(ctx context.Context, handler discordclient.EventHandler) error {
+	return c.serve(ctx, handler)
+}
+
+func TestStartupRepairRecoversGapWhileCaptureContinues(t *testing.T) {
+	for _, interval := range []time.Duration{0, 6 * time.Hour} {
+		for _, liveFirst := range []bool{false, true} {
+			t.Run(fmt.Sprintf("%s/live-first=%t", interval, liveFirst), func(t *testing.T) {
+				ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
+				defer cancel()
+				s, err := store.Open(ctx, filepath.Join(t.TempDir(), "archive.db"))
+				require.NoError(t, err)
+				defer func() { _ = s.Close() }()
+				require.NoError(t, s.UpsertMessage(ctx, store.MessageRecord{ID: "1", GuildID: "g1", ChannelID: "c1", Content: "before restart", NormalizedContent: "before restart"}))
+				require.NoError(t, s.SetSyncState(ctx, channelLatestScope("c1"), "1"))
+				blocked := make(chan struct{})
+				client := &startupTailClient{fakeClient: &fakeClient{
+					guilds:         []*discordgo.UserGuild{{ID: "g1", Name: "Guild"}},
+					guildByID:      map[string]*discordgo.Guild{"g1": {ID: "g1", Name: "Guild"}},
+					channels:       map[string][]*discordgo.Channel{"g1": {{ID: "c1", GuildID: "g1", Type: discordgo.ChannelTypeGuildText, LastMessageID: "11"}}},
+					messages:       map[string][]*discordgo.Message{"c1": {{ID: "10", GuildID: "g1", ChannelID: "c1", Content: "offline message", Timestamp: time.Now(), Author: &discordgo.User{ID: "u1"}}}},
+					messageBlocks:  map[string]chan struct{}{"c1": blocked},
+					messageStarted: make(chan string, 1),
+				}}
+				captured := make(chan struct{})
+				client.serve = func(ctx context.Context, handler discordclient.EventHandler) error {
+					capture := func() error {
+						if err := handler.OnMessageCreate(ctx, &discordgo.Message{ID: "11", GuildID: "g1", ChannelID: "c1", Content: "live message", Timestamp: time.Now(), Author: &discordgo.User{ID: "u1"}}); err != nil {
+							return err
+						}
+						close(captured)
+						return nil
+					}
+					if liveFirst {
+						if err := capture(); err != nil {
+							return err
+						}
+					}
+					if err := handler.(discordclient.TailReadyHandler).OnTailReady(ctx); err != nil {
+						return err
+					}
+					select {
+					case <-client.messageStarted:
+					case <-ctx.Done():
+						return ctx.Err()
+					}
+					// REST is blocked, but the connected Gateway still writes events.
+					if !liveFirst {
+						if err := capture(); err != nil {
+							return err
+						}
+					}
+					close(blocked)
+					<-ctx.Done()
+					return nil
+				}
+				svc := New(client, s, nil)
+				svc.SetTailRepairOnStart(true)
+				svc.SetTailEmbeddings(true)
+				done := make(chan error, 1)
+				go func() { defer close(done); done <- svc.RunTail(ctx, []string{"g1"}, interval) }()
+				defer func() { cancel(); <-done }()
+				select {
+				case <-captured:
+				case <-ctx.Done():
+					t.Fatal("capture or immediate startup repair did not progress")
+				}
+				require.Eventually(t, func() bool {
+					var count int
+					_ = s.DB().QueryRowContext(ctx, `select count(*) from messages where id in ('10','11')`).Scan(&count)
+					return count == 2
+				}, 2*time.Second, 10*time.Millisecond)
+				cancel()
+				require.NoError(t, <-done)
+				var queued int
+				require.NoError(t, s.DB().QueryRowContext(t.Context(), `select count(*) from embedding_jobs where message_id in ('10','11')`).Scan(&queued))
+				require.Equal(t, 2, queued)
+			})
+		}
+	}
+}
+
+func TestStartupRepairWaitsForReadyAndJoinsOnShutdown(t *testing.T) {
+	ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
+	defer cancel()
+	s, err := store.Open(ctx, filepath.Join(t.TempDir(), "archive.db"))
+	require.NoError(t, err)
+	defer func() { _ = s.Close() }()
+	connected, allowReady, started, joined := make(chan struct{}), make(chan struct{}), make(chan struct{}), make(chan struct{})
+	client := &startupTailClient{fakeClient: &fakeClient{}, serve: func(ctx context.Context, h discordclient.EventHandler) error {
+		close(connected)
+		select {
+		case <-allowReady:
+		case <-ctx.Done():
+			return nil
+		}
+		if err := h.(discordclient.TailReadyHandler).OnTailReady(ctx); err != nil {
+			return err
+		}
+		<-ctx.Done()
+		return nil
+	}}
+	svc := New(client, s, nil)
+	svc.SetTailRepairOnStart(true)
+	var ready atomic.Bool
+	svc.SetTailReadyCallback(func(context.Context) error { ready.Store(true); return nil })
+	svc.tailRepair = func(ctx context.Context, opts SyncOptions) (SyncStats, error) {
+		if !ready.Load() {
+			t.Error("repair started before ownership-ready callback")
+		}
+		close(started)
+		<-ctx.Done()
+		close(joined)
+		return SyncStats{}, ctx.Err()
+	}
+	done := make(chan error, 1)
+	go func() { done <- svc.RunTail(ctx, nil, time.Millisecond) }()
+	<-connected
+	select {
+	case <-started:
+		t.Fatal("repair ran before Gateway ready")
+	case <-time.After(20 * time.Millisecond):
+	}
+	close(allowReady)
+	select {
+	case <-started:
+	case <-ctx.Done():
+		t.Fatal("startup repair did not start")
+	}
+	cancel()
+	require.NoError(t, <-done)
+	select {
+	case <-joined:
+	default:
+		t.Fatal("repair was not joined before return")
+	}
+}
+
+func TestStartupRepairCursorSurvivesInterruptedOwner(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "archive.db")
+	s, err := store.Open(t.Context(), path)
+	require.NoError(t, err)
+	require.NoError(t, s.SetSyncState(t.Context(), channelLatestScope("c1"), "1"))
+	handler := &tailHandler{store: s, preserveHistoryCursor: true}
+	message := &discordgo.Message{ID: "11", GuildID: "g1", ChannelID: "c1", Content: "live before repair", Timestamp: time.Now(), Author: &discordgo.User{ID: "u1"}}
+	require.NoError(t, handler.OnMessageCreate(t.Context(), message))
+	cursor, err := s.GetSyncState(t.Context(), channelLatestScope("c1"))
+	require.NoError(t, err)
+	require.Equal(t, "1", cursor)
+	freshness, err := s.GetSyncState(t.Context(), "tail:last_event")
+	require.NoError(t, err)
+	require.Equal(t, "11", freshness)
+	// The first owner exits before REST repair can persist anything.
+	require.NoError(t, s.Close())
+	s, err = store.Open(t.Context(), path)
+	require.NoError(t, err)
+	defer func() { _ = s.Close() }()
+	ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
+	defer cancel()
+	client := &startupTailClient{fakeClient: &fakeClient{
+		guilds:    []*discordgo.UserGuild{{ID: "g1", Name: "Guild"}},
+		guildByID: map[string]*discordgo.Guild{"g1": {ID: "g1", Name: "Guild"}},
+		channels:  map[string][]*discordgo.Channel{"g1": {{ID: "c1", GuildID: "g1", Type: discordgo.ChannelTypeGuildText, LastMessageID: "11"}}},
+		messages: map[string][]*discordgo.Message{"c1": {
+			message,
+			{ID: "10", GuildID: "g1", ChannelID: "c1", Content: "missed offline", Timestamp: time.Now(), Author: &discordgo.User{ID: "u1"}},
+		}},
+	}, serve: func(ctx context.Context, h discordclient.EventHandler) error {
+		if err := h.(discordclient.TailReadyHandler).OnTailReady(ctx); err != nil {
+			return err
+		}
+		<-ctx.Done()
+		return nil
+	}}
+	svc := New(client, s, nil)
+	svc.SetTailRepairOnStart(true)
+	done := make(chan error, 1)
+	go func() { defer close(done); done <- svc.RunTail(ctx, []string{"g1"}, 6*time.Hour) }()
+	defer func() { cancel(); <-done }()
+	require.Eventually(t, func() bool {
+		var count int
+		_ = s.DB().QueryRowContext(ctx, `select count(*) from messages where id in ('10','11')`).Scan(&count)
+		cursor, _ := s.GetSyncState(ctx, channelLatestScope("c1"))
+		return count == 2 && cursor == "11"
+	}, 2*time.Second, 10*time.Millisecond)
+	// Later live events still cannot claim unverified history coverage.
+	handler.store = s
+	later := *message
+	later.ID = "12"
+	require.NoError(t, handler.OnMessageCreate(ctx, &later))
+	cursor, err = s.GetSyncState(ctx, channelLatestScope("c1"))
+	require.NoError(t, err)
+	require.Equal(t, "11", cursor)
+	cancel()
+	require.NoError(t, <-done)
+}

internal/syncer/syncer.go

@@ -50,6 +50,7 @@ type Syncer struct {
 	tailRepairOffsetMu    sync.RWMutex
 	tailRepairOffset      time.Duration
 	tailEmbeddings        bool
+	tailRepairOnStart     bool
 	channelExclusions     channelExclusions
 }
 
@@ -75,6 +76,10 @@ func (s *Syncer) SetTailReadyCallback(fn func(context.Context) error) {
 	s.tailReady = fn
 }
 
+func (s *Syncer) SetTailRepairOnStart(enabled bool) {
+	s.tailRepairOnStart = enabled
+}
+
 func (s *Syncer) SetChannelExclusions(channelIDs, channelKinds []string) {
 	s.channelExclusions.ids = normalizedStringSet(channelIDs, false)
 	s.channelExclusions.kinds = normalizedStringSet(channelKinds, true)