fix: delete data down to the remaining watermark on reorg

cmd/shovel/main.go

@@ -51,6 +51,7 @@ func main() {
 		profile     string
 		version     bool
 		verbose     bool
+		stallExit   time.Duration
 	)
 	flag.StringVar(&cfile, "config", "", "task config file")
 	flag.BoolVar(&printSchema, "print-schema", false, "print schema and exit")
@@ -60,6 +61,8 @@ func main() {
 	flag.StringVar(&profile, "profile", "", "run profile after indexing")
 	flag.BoolVar(&version, "version", false, "version")
 	flag.BoolVar(&verbose, "v", false, "verbose logging")
+	flag.DurationVar(&stallExit, "stall-exit", 0,
+		"exit when a task's watermark stalls this long while siblings advance (0 disables)")
 
 	flag.Parse()
 
@@ -189,6 +192,19 @@ func main() {
 		}
 	}()
 
+	if stallExit > 0 {
+		go mgr.WatchStalls(ctx, time.Minute, stallExit, func(stalled []string) {
+			slog.ErrorContext(ctx, "task-stalled",
+				"tasks", strings.Join(stalled, ","),
+				"stall_exit", stallExit.String(),
+			)
+			// The frozen goroutine's stack is the diagnosis; capture it before
+			// the supervisor replaces the process.
+			pprof.Lookup("goroutine").WriteTo(os.Stderr, 2)
+			os.Exit(70)
+		})
+	}
+
 	ec := make(chan error)
 	go mgr.Run(ec)
 	if err := <-ec; err != nil {

shovel/stall.go

@@ -0,0 +1,142 @@
+package shovel
+
+import (
+	"context"
+	"log/slog"
+	"sort"
+	"time"
+)
+
+// stallWatch tracks per-task watermark progress across polls. A task is stalled
+// when its watermark has not advanced for at least stallAfter while some other
+// task on the same source has advanced since — proof the chain head, the RPC,
+// and the database were all healthy while the task sat still. A source-wide
+// freeze (an RPC outage) is deliberately not a stall: restarting would not help
+// and would loop.
+type stallWatch struct {
+	stallAfter time.Duration
+	seen       map[string]stallSeen // task key -> last distinct watermark
+	srcAdvance map[string]time.Time // src -> last time any of its tasks advanced
+}
+
+type stallSeen struct {
+	num   uint64
+	since time.Time
+}
+
+type taskTip struct {
+	src, ig string
+	num     uint64
+}
+
+func newStallWatch(stallAfter time.Duration) *stallWatch {
+	return &stallWatch{
+		stallAfter: stallAfter,
+		seen:       map[string]stallSeen{},
+		srcAdvance: map[string]time.Time{},
+	}
+}
+
+// check records the current tips and returns the keys (src/ig) of stalled
+// tasks. First sight of a task seeds its state and is not an advance.
+func (w *stallWatch) check(now time.Time, tips []taskTip) []string {
+	for _, t := range tips {
+		k := t.src + "/" + t.ig
+		s, ok := w.seen[k]
+		if !ok || t.num != s.num {
+			w.seen[k] = stallSeen{num: t.num, since: now}
+			if ok {
+				w.srcAdvance[t.src] = now
+			}
+		}
+	}
+	var stalled []string
+	for _, t := range tips {
+		k := t.src + "/" + t.ig
+		s := w.seen[k]
+		if now.Sub(s.since) < w.stallAfter {
+			continue
+		}
+		if adv, ok := w.srcAdvance[t.src]; !ok || !adv.After(s.since) {
+			continue
+		}
+		stalled = append(stalled, k)
+	}
+	sort.Strings(stalled)
+	return stalled
+}
+
+// WatchStalls polls task watermarks every interval and calls onStall with the
+// stalled tasks per [stallWatch.check]. Tasks that are disabled, gate on
+// dependencies, or have a configured stop are exempt.
+//
+// A converge loop can freeze without an error line — its goroutine parked or
+// gone — and new-task only happens at process start, so no in-process retry
+// can revive it. The intended onStall dumps goroutine stacks and exits,
+// letting the supervisor (docker restart policy, systemd) replace the process.
+func (tm *Manager) WatchStalls(ctx context.Context, interval, stallAfter time.Duration, onStall func(stalled []string)) {
+	w := newStallWatch(stallAfter)
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+		}
+		tips, err := tm.taskTips(ctx)
+		if err != nil {
+			slog.ErrorContext(ctx, "stall-watch", "error", err)
+			continue
+		}
+		if stalled := w.check(time.Now(), tips); len(stalled) > 0 {
+			onStall(stalled)
+		}
+	}
+}
+
+// taskTips returns the current watermark for every stall-eligible task. A task
+// with no task_updates rows reports 0, so a task that dies before its first
+// write still counts as stalled once its siblings move.
+func (tm *Manager) taskTips(ctx context.Context) ([]taskTip, error) {
+	igs, err := tm.conf.AllIntegrations(ctx, tm.pgp)
+	if err != nil {
+		return nil, err
+	}
+	const q = `
+		select src_name, ig_name, max(num)
+		from shovel.task_updates
+		group by 1, 2
+	`
+	rows, err := tm.pgp.Query(ctx, q)
+	if err != nil {
+		return nil, err
+	}
+	nums := map[string]uint64{}
+	for rows.Next() {
+		var (
+			src, ig string
+			num     uint64
+		)
+		if err := rows.Scan(&src, &ig, &num); err != nil {
+			return nil, err
+		}
+		nums[src+"/"+ig] = num
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	var tips []taskTip
+	for _, ig := range igs {
+		if !ig.Enabled || len(ig.Dependencies) > 0 {
+			continue
+		}
+		for _, src := range ig.Sources {
+			if src.Stop > 0 {
+				continue
+			}
+			tips = append(tips, taskTip{src: src.Name, ig: ig.Name, num: nums[src.Name+"/"+ig.Name]})
+		}
+	}
+	return tips, nil
+}

shovel/stall_test.go

@@ -0,0 +1,90 @@
+package shovel
+
+import (
+	"testing"
+	"time"
+
+	"kr.dev/diff"
+)
+
+func TestStallWatch(t *testing.T) {
+	var (
+		t0   = time.Unix(1000, 0)
+		tick = func(n int) time.Time { return t0.Add(time.Duration(n) * time.Minute) }
+		tips = func(a, b uint64) []taskTip {
+			return []taskTip{
+				{src: "m", ig: "one", num: a},
+				{src: "m", ig: "two", num: b},
+			}
+		}
+		w = newStallWatch(5 * time.Minute)
+	)
+	// seed. nothing can be stalled on first sight
+	diff.Test(t, t.Errorf, w.check(tick(0), tips(10, 10)), []string(nil))
+	// both advance every tick. healthy
+	for i := 1; i <= 10; i++ {
+		diff.Test(t, t.Errorf, w.check(tick(i), tips(10+uint64(i), 10+uint64(i))), []string(nil))
+	}
+	// "two" freezes at 20 while "one" keeps advancing. below threshold: quiet
+	for i := 11; i <= 14; i++ {
+		diff.Test(t, t.Errorf, w.check(tick(i), tips(10+uint64(i), 20)), []string(nil))
+	}
+	// threshold reached and a sibling advanced since the freeze: stalled
+	diff.Test(t, t.Errorf, w.check(tick(15), tips(25, 20)), []string{"m/two"})
+}
+
+func TestStallWatch_SourceWideFreeze(t *testing.T) {
+	var (
+		t0   = time.Unix(1000, 0)
+		tick = func(n int) time.Time { return t0.Add(time.Duration(n) * time.Minute) }
+		w    = newStallWatch(5 * time.Minute)
+		tips = []taskTip{
+			{src: "m", ig: "one", num: 10},
+			{src: "m", ig: "two", num: 10},
+		}
+	)
+	// every task on the source frozen (rpc outage): never stalled
+	for i := 0; i <= 30; i++ {
+		diff.Test(t, t.Errorf, w.check(tick(i), tips), []string(nil))
+	}
+}
+
+func TestStallWatch_NeverWrote(t *testing.T) {
+	var (
+		t0   = time.Unix(1000, 0)
+		tick = func(n int) time.Time { return t0.Add(time.Duration(n) * time.Minute) }
+		w    = newStallWatch(5 * time.Minute)
+		tips = func(a uint64) []taskTip {
+			return []taskTip{
+				{src: "m", ig: "one", num: a},
+				{src: "m", ig: "two", num: 0}, // no task_updates row ever
+			}
+		}
+	)
+	diff.Test(t, t.Errorf, w.check(tick(0), tips(10)), []string(nil))
+	for i := 1; i <= 4; i++ {
+		diff.Test(t, t.Errorf, w.check(tick(i), tips(10+uint64(i))), []string(nil))
+	}
+	diff.Test(t, t.Errorf, w.check(tick(5), tips(15)), []string{"m/two"})
+}
+
+func TestStallWatch_RecoversOnAdvance(t *testing.T) {
+	var (
+		t0   = time.Unix(1000, 0)
+		tick = func(n int) time.Time { return t0.Add(time.Duration(n) * time.Minute) }
+		w    = newStallWatch(5 * time.Minute)
+		tips = func(a, b uint64) []taskTip {
+			return []taskTip{
+				{src: "m", ig: "one", num: a},
+				{src: "m", ig: "two", num: b},
+			}
+		}
+	)
+	diff.Test(t, t.Errorf, w.check(tick(0), tips(10, 10)), []string(nil))
+	for i := 1; i <= 4; i++ {
+		diff.Test(t, t.Errorf, w.check(tick(i), tips(10+uint64(i), 10)), []string(nil))
+	}
+	// frozen task catches up right at the threshold: healthy again
+	diff.Test(t, t.Errorf, w.check(tick(5), tips(15, 15)), []string(nil))
+	diff.Test(t, t.Errorf, w.check(tick(6), tips(16, 16)), []string(nil))
+}

shovel/task.go

@@ -231,6 +231,17 @@ func (t *Task) update(
 	return err
 }
 
+// Delete removes the task updates at or above n and the destination's data
+// for every block those updates covered.
+//
+// [Task.update] writes one shovel.task_updates row per converge batch and sets
+// num to the batch's LAST block, so a single update covers (previous update,
+// num]. Deleting data at >= n only would therefore keep the deleted batch's
+// canonical prefix -- blocks below n that no remaining update accounts for.
+// Converge resumes from the highest remaining update and requests those blocks
+// again, and since [Destination.Insert] uses COPY (no on conflict), the retry
+// fails against the destination's unique index forever: the task stops making
+// progress until the leftover rows are deleted by hand.
 func (t *Task) Delete(pg wpg.Conn, n uint64) error {
 	const q = `
 		delete from shovel.task_updates
@@ -242,17 +253,51 @@ func (t *Task) Delete(pg wpg.Conn, n uint64) error {
 	if err != nil {
 		return fmt.Errorf("deleting block from task table: %w", err)
 	}
-	err = t.dests[0].Delete(t.ctx, pg, n)
+	dn, err := t.resume(pg, n)
+	if err != nil {
+		return err
+	}
+	err = t.dests[0].Delete(t.ctx, pg, dn)
 	if err != nil {
 		return fmt.Errorf("deleting block: %w", err)
 	}
 	slog.InfoContext(t.ctx, "task-delete",
 		"n", n,
+		"data_n", dn,
 		"task_updates", cmd.RowsAffected(),
 	)
 	return nil
 }
 
+// resume reports the first block Converge will request once the task updates
+// at or above n are gone: one past the highest remaining update. It is never
+// greater than n, since no update at or above n survives the delete.
+//
+// With no updates left the task restarts at its configured start, or -- absent
+// one -- at the chain head, which is above anything already stored. Both cases
+// fall back to n rather than deleting the destination's whole history.
+func (t *Task) resume(pg wpg.Conn, n uint64) (uint64, error) {
+	const q = `
+		select coalesce(max(num) + 1, 0)
+		from shovel.task_updates
+		where src_name = $1
+		and ig_name = $2
+	`
+	var num uint64
+	err := pg.QueryRow(t.ctx, q, t.srcName, t.destConfig.Name).Scan(&num)
+	if err != nil {
+		return 0, fmt.Errorf("querying remaining task updates: %w", err)
+	}
+	switch {
+	case num > 0:
+		return num, nil
+	case t.start > 0:
+		return t.start, nil
+	default:
+		return n, nil
+	}
+}
+
 func (t *Task) latestDependency(pg wpg.Conn) (uint64, []byte, error) {
 	const q = `
 		with latest as (