diff --git a/internal/telemetry/compaction_debt.go b/internal/telemetry/compaction_debt.go
index a83be249..7f86943f 100644
--- a/internal/telemetry/compaction_debt.go
+++ b/internal/telemetry/compaction_debt.go
@@ -78,6 +78,33 @@ func (p *ParquetStore) setBatchLocked(id string, batch *storedBatch) {
 		close(p.debtChanged)
 		p.debtChanged = nil
 	}
+	if p.compactionDebt >= CompactionDebtLimit && p.debtLimitReached != nil {
+		close(p.debtLimitReached)
+		p.debtLimitReached = nil
+	}
+}
+
+// WaitForCompactionLimit wakes higher-generation work when ingest needs the
+// compactor. Registration and the threshold check share the catalog lock, so
+// debt cannot cross the limit between the check and subscribing. No lock is
+// held while waiting, and caller cancellation releases the waiter.
+func (p *ParquetStore) WaitForCompactionLimit(ctx context.Context) error {
+	p.mu.Lock()
+	if p.compactionDebt >= CompactionDebtLimit {
+		p.mu.Unlock()
+		return nil
+	}
+	if p.debtLimitReached == nil {
+		p.debtLimitReached = make(chan struct{})
+	}
+	reached := p.debtLimitReached
+	p.mu.Unlock()
+	select {
+	case <-reached:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
 }
 
 // WaitForCompaction releases the catalog lock before sleeping and wakes on a
diff --git a/internal/telemetry/compaction_debt_test.go b/internal/telemetry/compaction_debt_test.go
index d5c28ca8..05ffa280 100644
--- a/internal/telemetry/compaction_debt_test.go
+++ b/internal/telemetry/compaction_debt_test.go
@@ -2,11 +2,65 @@ package telemetry
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"testing"
 	"time"
 )
 
+func TestCompactionLimitNotificationDoesNotMissCrossings(t *testing.T) {
+	p := &ParquetStore{batches: make(map[string]*storedBatch)}
+	for i := range CompactionDebtLimit + CompactionBatchLimit - 2 {
+		p.installBatch(&storedBatch{metadata: BatchMetadata{ID: fmt.Sprint(i), Spans: 1, MaxIngestedNanos: 1}})
+	}
+	for cycle := range 2 {
+		ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
+		done := make(chan error, 2)
+		for range 2 {
+			go func() { done <- p.WaitForCompactionLimit(ctx) }()
+		}
+		until := time.Now().Add(time.Second)
+		for {
+			p.mu.RLock()
+			registered := p.debtLimitReached != nil
+			p.mu.RUnlock()
+			if registered {
+				break
+			}
+			if time.Now().After(until) {
+				cancel()
+				t.Fatal("limit waiter not registered")
+			}
+			time.Sleep(time.Millisecond)
+		}
+		p.installBatch(&storedBatch{metadata: BatchMetadata{ID: "crossing", Spans: 1, MaxIngestedNanos: 1}})
+		for range 2 {
+			if err := <-done; err != nil {
+				cancel()
+				t.Fatal(err)
+			}
+		}
+		// Arriving after the crossing also observes it, without waiting for
+		// another publication. Retiring it rearms the next crossing.
+		if err := p.WaitForCompactionLimit(ctx); err != nil {
+			cancel()
+			t.Fatal(err)
+		}
+		cancel()
+		p.mu.Lock()
+		p.setBatchLocked("crossing", nil)
+		p.mu.Unlock()
+		if p.CompactionDebt() != CompactionDebtLimit-1 {
+			t.Fatalf("cycle %d debt=%d", cycle, p.CompactionDebt())
+		}
+	}
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+	if err := p.WaitForCompactionLimit(ctx); !errors.Is(err, context.Canceled) {
+		t.Fatal(err)
+	}
+}
+
 func TestGenerationZeroInventoryTracksCatalogChanges(t *testing.T) {
 	p := &ParquetStore{batches: make(map[string]*storedBatch)}
 	add := func(id string, generation int) {
diff --git a/internal/telemetry/compaction_index_test.go b/internal/telemetry/compaction_index_test.go
index 425ee3c5..db2f90de 100644
--- a/internal/telemetry/compaction_index_test.go
+++ b/internal/telemetry/compaction_index_test.go
@@ -118,7 +118,9 @@ type cancelMergeContext struct {
 
 func (c *cancelMergeContext) Err() error {
 	c.checks++
-	if c.checks == 3 {
+	// Initial admission, input open and key priming now check cancellation too.
+	// Cancel at the next row-group boundary, after the first group was written.
+	if c.checks == 5 {
 		c.cancel()
 	}
 	return c.Context.Err()
diff --git a/internal/telemetry/parquet.go b/internal/telemetry/parquet.go
index 4ad62f44..332b0f4b 100644
--- a/internal/telemetry/parquet.go
+++ b/internal/telemetry/parquet.go
@@ -94,6 +94,7 @@ type ParquetStore struct {
 	generation0ByDay   map[int64]int
 	compactionDebt     int
 	debtChanged        chan struct{}
+	debtLimitReached   chan struct{}
 }
 
 type ParquetStats struct {
diff --git a/internal/telemetry/parquet_merge.go b/internal/telemetry/parquet_merge.go
index 98c48f63..0e298ee1 100644
--- a/internal/telemetry/parquet_merge.go
+++ b/internal/telemetry/parquet_merge.go
@@ -1,6 +1,7 @@
 package telemetry
 
 import (
+	"bytes"
 	"container/heap"
 	"context"
 	"errors"
@@ -37,6 +38,9 @@ func mergeTypedParquet[T any](ctx context.Context, inputs []string, output strin
 	schema := telemetrySchema[T]()
 	var groups []parquet.RowGroup
 	for _, input := range inputs {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
 		f, e := os.Open(input)
 		if e != nil {
 			return e
@@ -96,18 +100,25 @@ func mergeTypedParquet[T any](ctx context.Context, inputs []string, output strin
 			}
 		}()
 		for i, group := range groups {
-			c := &mergeKeyCursor{input: i, reader: parquet.NewGenericRowGroupReader[mergeSpanKey](group)}
+			c := &mergeKeyCursor{input: i, remaining: group.NumRows()}
+			for k, name := range [...]string{"_trace_hash", "start_unix_nano", "span_id"} {
+				leaf, _ := schema.Lookup(name)
+				c.readers[k] = parquet.NewColumnChunkValueReader(group.ColumnChunks()[leaf.ColumnIndex])
+			}
 			// Keep empty/exhausted cursors in a separate lifetime list below.
 			keys = append(keys, c)
 		}
 		allKeys := append(mergeKeyHeap(nil), keys...)
 		defer func() {
 			for _, c := range allKeys {
-				err = errors.Join(err, c.reader.Close())
+				err = errors.Join(err, c.close())
 			}
 		}()
 		keys = keys[:0]
 		for _, c := range allKeys {
+			if err := ctx.Err(); err != nil {
+				return err
+			}
 			has, e := c.next()
 			if e != nil {
 				return e
@@ -285,42 +296,60 @@ func (c *mergeColumn) write(ctx context.Context, writer *parquet.ColumnWriter, r
 	return writer.Flush()
 }
 
+// Key values borrow their column reader's current page. Each reader advances
+// only after its current key has been consumed; no per-span string or row copy
+// is needed, and the three columns may have independent page boundaries.
 type mergeSpanKey struct {
-	Hash   uint64 `parquet:"_trace_hash"`
-	Start  int64  `parquet:"start_unix_nano"`
-	SpanID string `parquet:"span_id"`
+	Hash   uint64
+	Start  int64
+	SpanID []byte
 }
 type mergeKeyCursor struct {
-	input  int
-	reader *parquet.GenericReader[mergeSpanKey]
-	buffer [256]mergeSpanKey
-	pos, n int
-	eof    bool
+	input     int
+	readers   [3]parquet.ColumnChunkValueReader
+	buffer    [3][256]parquet.Value
+	pos, n    [3]int
+	remaining int64
+	key       mergeSpanKey
 }
 
-func (c *mergeKeyCursor) current() mergeSpanKey { return c.buffer[c.pos-1] }
-func (c *mergeKeyCursor) next() (bool, error) {
-	if c.pos == c.n {
-		if c.eof {
-			return false, nil
+func (c *mergeKeyCursor) close() (err error) {
+	for i, r := range c.readers {
+		if r != nil {
+			err = errors.Join(err, r.Close())
+			c.readers[i] = nil
 		}
-		clear(c.buffer[:])
-		n, e := c.reader.Read(c.buffer[:])
-		c.pos, c.n = 0, n
-		if e != nil {
-			if e != io.EOF {
+	}
+	clear(c.buffer[:])
+	c.key = mergeSpanKey{}
+	return err
+}
+func (c *mergeKeyCursor) current() mergeSpanKey { return c.key }
+func (c *mergeKeyCursor) next() (bool, error) {
+	if c.remaining == 0 {
+		return false, c.close()
+	}
+	var values [3]parquet.Value
+	for k, r := range c.readers {
+		if c.pos[k] == c.n[k] {
+			clear(c.buffer[k][:])
+			n, e := r.ReadValues(c.buffer[k][:])
+			c.pos[k], c.n[k] = 0, n
+			if e != nil && e != io.EOF {
 				return false, e
 			}
-			c.eof = true
-		}
-		if n == 0 {
-			if c.eof {
-				return false, nil
+			if n == 0 {
+				if e == io.EOF {
+					return false, io.ErrUnexpectedEOF
+				}
+				return false, io.ErrNoProgress
 			}
-			return false, io.ErrNoProgress
 		}
+		values[k] = c.buffer[k][c.pos[k]]
+		c.pos[k]++
 	}
-	c.pos++
+	c.key = mergeSpanKey{Hash: values[0].Uint64(), Start: values[1].Int64(), SpanID: values[2].ByteArray()}
+	c.remaining--
 	return true, nil
 }
 
@@ -335,8 +364,8 @@ func (h mergeKeyHeap) Less(i, j int) bool {
 	if a.Start != b.Start {
 		return a.Start < b.Start
 	}
-	if a.SpanID != b.SpanID {
-		return a.SpanID < b.SpanID
+	if order := bytes.Compare(a.SpanID, b.SpanID); order != 0 {
+		return order < 0
 	}
 	return h[i].input < h[j].input
 }
diff --git a/internal/telemetry/store/backpressure_test.go b/internal/telemetry/store/backpressure_test.go
index e34c9231..c936101e 100644
--- a/internal/telemetry/store/backpressure_test.go
+++ b/internal/telemetry/store/backpressure_test.go
@@ -120,7 +120,7 @@ func TestCompactionPressureDrainsGenerationZeroBeforeHigherGenerations(t *testin
 			}
 			debt := telemetry.CompactionDebtLimit
 			for pass := 0; debt >= telemetry.CompactionDebtLimit/2; pass++ {
-				selected := selectCompactionBatches(batches, 128, debt)
+				selected := selectCompactionBatchesWithHigherCap(batches, 128, 128, debt)
 				if len(selected) < 2 {
 					t.Fatal("positive debt has no selection")
 				}
@@ -146,7 +146,7 @@ func TestCompactionPressureDrainsGenerationZeroBeforeHigherGenerations(t *testin
 				batches = next
 			}
 			if strings.HasPrefix(mode, "higher-") {
-				selected := selectCompactionBatches(batches, 128, debt)
+				selected := selectCompactionBatchesWithHigherCap(batches, 128, 128, debt)
 				if len(selected) < 2 || selected[0].Generation == 0 {
 					t.Fatalf("higher generations did not resume at debt=%d: %v", debt, selected)
 				}
@@ -160,14 +160,14 @@ func TestCompactionPressureDrainsGenerationZeroBeforeHigherGenerations(t *testin
 		small = append(small, telemetry.BatchMetadata{ID: fmt.Sprint(i), Spans: 1, MaxIngestedNanos: 1})
 	}
 	small = append(small, telemetry.BatchMetadata{ID: "today", Spans: 1, MaxIngestedNanos: int64(24*time.Hour) + 1})
-	if len(selectCompactionBatches(small, 128, 0)) != 8 {
+	if len(selectCompactionBatchesWithHigherCap(small, 128, 128, 0)) != 8 {
 		t.Fatal("finished eight-input group stranded")
 	}
 	var terminal []telemetry.BatchMetadata
 	for i := 0; i < 1024; i++ {
 		terminal = append(terminal, telemetry.BatchMetadata{ID: fmt.Sprint(i), Spans: maxCompactionRows, MaxIngestedNanos: 1})
 	}
-	if len(selectCompactionBatches(terminal, 128, 0)) != 0 {
+	if len(selectCompactionBatchesWithHigherCap(terminal, 128, 128, 0)) != 0 {
 		t.Fatal("terminal counterexample changed")
 	}
 }
diff --git a/internal/telemetry/store/compaction.go b/internal/telemetry/store/compaction.go
index 201d93a3..0ede5033 100644
--- a/internal/telemetry/store/compaction.go
+++ b/internal/telemetry/store/compaction.go
@@ -135,6 +135,25 @@ func (r *Repository) CompactParquet(ctx context.Context, publisher ParquetPublis
 	}
 	r.compactionMu.Lock()
 	defer r.compactionMu.Unlock()
+	for {
+		count, err := r.compactParquet(ctx, publisher, maxBatches)
+		if !errors.Is(err, errCompactionPreempted) {
+			return count, err
+		}
+		// The interrupted output was never prepared or marked. Its workers and
+		// staging cleanup have finished. The higher-generation retry size is
+		// retained separately while generation 0 retires debt and clears its
+		// own attempt record. Reselect immediately, prioritizing generation 0.
+		if err := clearCompactionAttempt(r.root); err != nil {
+			return 0, err
+		}
+	}
+}
+
+func (r *Repository) compactParquet(ctx context.Context, publisher ParquetPublisher, maxBatches int) (int, error) {
+	if err := ctx.Err(); err != nil {
+		return 0, err
+	}
 	markerPath := filepath.Join(r.root, "COMPACTION.json")
 	if exists, err := pathExists(markerPath); err != nil {
 		return 0, err
@@ -145,7 +164,9 @@ func (r *Repository) CompactParquet(ctx context.Context, publisher ParquetPublis
 	}
 	// A pass that was killed mid-merge left a record of how much it tried to
 	// take. Take less.
-	selected := selectCompactionBatches(r.Parquet.BatchMetadata(), compactionBatchCap(maxBatches, readCompactionAttempt(r.root)), r.Parquet.CompactionDebt())
+	batchCap := compactionBatchCap(maxBatches, readCompactionAttempt(r.root))
+	selected := selectCompactionBatchesWithHigherCap(r.Parquet.BatchMetadata(), batchCap,
+		compactionBatchCap(batchCap, r.preemptedCompactionInputs), r.Parquet.CompactionDebt())
 	if len(selected) < minCompactionMergeInputs {
 		return 0, nil
 	}
@@ -210,30 +231,33 @@ func (r *Repository) CompactParquet(ctx context.Context, publisher ParquetPublis
 	if err := writeCompactionAttempt(r.root, len(selected)); err != nil {
 		return 0, err
 	}
-	group, mergeCtx := errgroup.WithContext(ctx)
-	for _, plan := range plans {
-		group.Go(func() error {
-			stats, err := r.Parquet.MergeParquet(mergeCtx, plan.signal, plan.inputs, plan.output)
-			if err != nil {
-				return fmt.Errorf("compact %s Parquet: %w", plan.signal, err)
-			}
-			if plan.signal == "spans" {
-				if stats.Rows != int64(marker.Output.Spans) {
-					return fmt.Errorf("compacted span count: got %d want %d", stats.Rows, marker.Output.Spans)
+	err := r.mergeCompaction(ctx, selected[0].Generation, len(selected), func(ctx context.Context) error {
+		group, mergeCtx := errgroup.WithContext(ctx)
+		for _, plan := range plans {
+			group.Go(func() error {
+				stats, err := r.Parquet.MergeParquet(mergeCtx, plan.signal, plan.inputs, plan.output)
+				if err != nil {
+					return fmt.Errorf("compact %s Parquet: %w", plan.signal, err)
 				}
-				marker.Output.MinSpanStartNanos = stats.MinSpanStartNanos
-				marker.Output.MaxSpanStartNanos = stats.MaxSpanStartNanos
-			}
-			if err := mergeCtx.Err(); err != nil {
-				return err
-			}
-			if err := syncFile(plan.output); err != nil {
-				return fmt.Errorf("sync compacted %s Parquet: %w", plan.signal, err)
-			}
-			return nil
-		})
-	}
-	if err := group.Wait(); err != nil {
+				if plan.signal == "spans" {
+					if stats.Rows != int64(marker.Output.Spans) {
+						return fmt.Errorf("compacted span count: got %d want %d", stats.Rows, marker.Output.Spans)
+					}
+					marker.Output.MinSpanStartNanos = stats.MinSpanStartNanos
+					marker.Output.MaxSpanStartNanos = stats.MaxSpanStartNanos
+				}
+				if err := mergeCtx.Err(); err != nil {
+					return err
+				}
+				if err := syncFile(plan.output); err != nil {
+					return fmt.Errorf("sync compacted %s Parquet: %w", plan.signal, err)
+				}
+				return nil
+			})
+		}
+		return group.Wait()
+	})
+	if err != nil {
 		return 0, err
 	}
 	if err := r.Parquet.PrepareReplacement(stage, marker.Output); err != nil {
@@ -259,7 +283,37 @@ func (r *Repository) CompactParquet(ctx context.Context, publisher ParquetPublis
 	return len(selected), nil
 }
 
-func selectCompactionBatches(batches []telemetry.BatchMetadata, maxBatches, debt int) []telemetry.BatchMetadata {
+var errCompactionPreempted = errors.New("higher-generation compaction yielded to ingest debt")
+
+// Only unprepared higher-generation work is cancellable by debt. Join the
+// watcher and all merge workers before touching preparation or durable markers.
+// Generation 0 and minimum pairs retain the caller context: debt can delay a
+// higher generation only until its attempts halve to a pair. The retry size is
+// process-local; after a restart the durable interrupted-attempt rule still
+// shrinks unfinished work. Called with compactionMu held.
+func (r *Repository) mergeCompaction(ctx context.Context, generation uint32, inputs int, merge func(context.Context) error) error {
+	if generation == 0 || inputs <= minCompactionMergeInputs {
+		return merge(ctx)
+	}
+	mergeCtx, cancel := context.WithCancelCause(ctx)
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		if r.Parquet.WaitForCompactionLimit(mergeCtx) == nil {
+			cancel(errCompactionPreempted)
+		}
+	}()
+	err := merge(mergeCtx)
+	cancel(nil)
+	<-done
+	if (err == nil || errors.Is(err, context.Canceled)) && errors.Is(context.Cause(mergeCtx), errCompactionPreempted) && ctx.Err() == nil {
+		r.preemptedCompactionInputs = inputs
+		return errCompactionPreempted
+	}
+	return err
+}
+
+func selectCompactionBatchesWithHigherCap(batches []telemetry.BatchMetadata, maxBatches, higherMaxBatches, debt int) []telemetry.BatchMetadata {
 	// minCompactionInputs is a "worth the effort" policy and belongs to the
 	// caller, which applies it to the ceiling it was asked for. Here the floor
 	// is only what a merge needs: after an interrupted pass the ceiling is
@@ -292,11 +346,15 @@ func selectCompactionBatches(batches []telemetry.BatchMetadata, maxBatches, debt
 	var selected []telemetry.BatchMetadata
 	found := false
 	for key, group := range groups {
+		batchCap := maxBatches
+		if key.generation > 0 {
+			batchCap = min(batchCap, higherMaxBatches)
+		}
 		rowLimit := int64(maxCompactionRows)
 		if pressure && key.generation == 0 {
 			rowLimit = telemetry.CompactionPressureMaxRows
 		}
-		candidate := selectBoundedCompactionGroup(group, maxBatches, rowLimit, key.day < newestDay)
+		candidate := selectBoundedCompactionGroup(group, batchCap, rowLimit, key.day < newestDay)
 		if len(candidate) < 2 {
 			continue
 		}
@@ -424,8 +482,8 @@ func compactionBatchRows(batch telemetry.BatchMetadata) int64 {
 }
 
 // CompactParquetPass starts complete compactions until the phase budget
-// expires. An in-flight merge keeps the caller context so slow, valid work
-// commits instead of restarting the same input group on every pass.
+// expires. The phase budget never cancels an in-flight merge; only blocking
+// ingest debt preempts unprepared higher-generation work.
 func (r *Repository) CompactParquetPass(ctx context.Context, publisher ParquetPublisher, maxBatches int, budget time.Duration) (int, error) {
 	if maxBatches <= 0 || budget <= 0 {
 		return 0, nil
@@ -494,6 +552,11 @@ func (r *Repository) completeCompaction(ctx context.Context, marker compactionMa
 	}); err != nil {
 		return err
 	}
+	// Recovery can finish publication too; only higher-generation outputs
+	// reset the debt-preemption size.
+	if marker.Output.Generation > 1 {
+		r.preemptedCompactionInputs = 0
+	}
 	markerPath := filepath.Join(r.root, "COMPACTION.json")
 	if err := os.Remove(markerPath); err != nil && !errors.Is(err, os.ErrNotExist) {
 		return err
diff --git a/internal/telemetry/store/compaction_preemption_test.go b/internal/telemetry/store/compaction_preemption_test.go
new file mode 100644
index 00000000..cb7a3322
--- /dev/null
+++ b/internal/telemetry/store/compaction_preemption_test.go
@@ -0,0 +1,216 @@
+package store
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/labstack/fanout/internal/telemetry"
+)
+
+func TestHigherGenerationMergeYieldsWithoutBlockingDebtRetirement(t *testing.T) {
+	r, err := Open(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer r.Close()
+	seed := Batch{ID: "seed", Logs: []telemetry.Log{{Body: "typed", TimeUnixNanos: 100}}}
+	if err := r.Commit(t.Context(), seed); err != nil {
+		t.Fatal(err)
+	}
+	// Real immutable files and catalog mutations cross the same threshold as
+	// ingest. Only one small log file is shared; no fake debt counters.
+	for i := 0; i < telemetry.CompactionDebtLimit+telemetry.CompactionBatchLimit-2; i++ {
+		id := fmt.Sprint("debt-", i)
+		stage := filepath.Join(r.root, "ready", id)
+		if err := os.MkdirAll(stage, 0755); err != nil {
+			t.Fatal(err)
+		}
+		if err := os.Link(filepath.Join(r.Parquet.BatchPath("seed"), "logs.parquet"), filepath.Join(stage, "logs.parquet")); err != nil {
+			t.Fatal(err)
+		}
+		m := telemetry.BatchMetadata{ID: id, Logs: 1, MinIngestedNanos: time.Now().UnixNano(), MaxIngestedNanos: time.Now().UnixNano()}
+		if err := r.Parquet.PrepareReplacement(stage, m); err != nil {
+			t.Fatal(err)
+		}
+		if err := r.Parquet.PublishReplacement(stage, m, nil, func(f func(context.Context) error) error { return f(t.Context()) }); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if r.CompactionDebt() != telemetry.CompactionDebtLimit {
+		t.Fatal(r.CompactionDebt())
+	}
+	ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
+	defer cancel()
+	err = r.mergeCompaction(ctx, 2, 8, func(ctx context.Context) error { <-ctx.Done(); return ctx.Err() })
+	if !errors.Is(err, errCompactionPreempted) {
+		t.Fatal(err)
+	}
+	if err := r.mergeCompaction(ctx, 0, 8, func(ctx context.Context) error { return ctx.Err() }); err != nil {
+		t.Fatalf("generation zero canceled by debt: %v", err)
+	}
+	if err := r.mergeCompaction(ctx, 2, 2, func(mergeCtx context.Context) error {
+		if mergeCtx != ctx {
+			t.Fatal("minimum pair is still cancellable by debt")
+		}
+		return mergeCtx.Err()
+	}); err != nil {
+		t.Fatalf("minimum pair canceled by debt: %v", err)
+	}
+	failure := errors.New("input failure")
+	if err := r.mergeCompaction(ctx, 2, 8, func(context.Context) error { return failure }); !errors.Is(err, failure) {
+		t.Fatalf("lost storage error: %v", err)
+	}
+	// No watcher holds compactionMu, the catalog mutex, or the publish gate.
+	// Ordinary compaction can retire debt immediately after the canceled work.
+	if n, err := r.CompactParquet(ctx, &testParquetCompactor{}, 128); err != nil || n != 128 {
+		t.Fatalf("retire: %d %v", n, err)
+	}
+	if _, err := r.WaitForCompaction(ctx); err != nil {
+		t.Fatal(err)
+	}
+	if err := r.RecoverParquet(ctx, &testParquetCompactor{}); err != nil {
+		t.Fatal(err)
+	}
+	if err := r.CleanupParquet(); err != nil {
+		t.Fatal(err)
+	}
+	rows := r.RowCount()
+	if want := uint64(telemetry.CompactionDebtLimit + telemetry.CompactionBatchLimit - 1); rows != want {
+		t.Fatalf("compaction lost rows: got %d want %d", rows, want)
+	}
+	if err := r.Close(); err != nil {
+		t.Fatal(err)
+	}
+	reopened, err := Open(r.root)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer reopened.Close()
+	if reopened.RowCount() != rows || reopened.CompactionDebt() != r.CompactionDebt() {
+		t.Fatal("recovery changed inventory")
+	}
+}
+
+func TestHigherGenerationMergeJoinsWatcherOnCompletionAndCallerCancellation(t *testing.T) {
+	r := &Repository{Parquet: &telemetry.ParquetStore{}}
+	for range 20 {
+		if err := r.mergeCompaction(t.Context(), 1, 8, func(context.Context) error { return nil }); err != nil {
+			t.Fatal(err)
+		}
+		ctx, cancel := context.WithCancel(t.Context())
+		err := r.mergeCompaction(ctx, 1, 8, func(ctx context.Context) error { cancel(); <-ctx.Done(); return ctx.Err() })
+		cancel()
+		if !errors.Is(err, context.Canceled) || errors.Is(err, errCompactionPreempted) {
+			t.Fatalf("caller cancellation: %v", err)
+		}
+	}
+}
+
+// Repeated hard-limit crossings must walk 8 -> 4 -> 2 even though successful
+// generation-0 passes run between them and clear the durable attempt record.
+func TestHigherGenerationPublishesAfterRepeatedDebtPreemption(t *testing.T) {
+	r, err := Open(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer r.Close()
+	ctx, cancel := context.WithTimeout(t.Context(), 3*time.Minute)
+	defer cancel()
+	if err := r.Commit(ctx, Batch{ID: "seed", Logs: []telemetry.Log{{Body: "progress", TimeUnixNanos: 100}}}); err != nil {
+		t.Fatal(err)
+	}
+	stamp := r.Parquet.BatchMetadata()[0].MaxIngestedNanos
+	sequence := 0
+	publishBatch := func(generation uint32) {
+		t.Helper()
+		sequence++
+		id := fmt.Sprintf("batch-%05d", sequence)
+		stage := filepath.Join(r.root, "ready", id)
+		if err := os.MkdirAll(stage, 0o755); err != nil {
+			t.Fatal(err)
+		}
+		if err := os.Link(filepath.Join(r.Parquet.BatchPath("seed"), "logs.parquet"), filepath.Join(stage, "logs.parquet")); err != nil {
+			t.Fatal(err)
+		}
+		m := telemetry.BatchMetadata{ID: id, Generation: generation, Logs: 1, MinIngestedNanos: stamp, MaxIngestedNanos: stamp}
+		if err := r.Parquet.PrepareReplacement(stage, m); err != nil {
+			t.Fatal(err)
+		}
+		if err := r.Parquet.PublishReplacement(stage, m, nil, func(f func(context.Context) error) error { return f(ctx) }); err != nil {
+			t.Fatal(err)
+		}
+	}
+	// Keep the shared immutable seed on a day with no mergeable peers.
+	stamp += int64(24 * time.Hour)
+	for range 8 {
+		publishBatch(1)
+	}
+	fillDebt := func(target int) {
+		t.Helper()
+		for r.CompactionDebt() < target {
+			publishBatch(0)
+		}
+	}
+	compactor := &testParquetCompactor{}
+	drainDebt := func() {
+		t.Helper()
+		for r.CompactionDebt() >= telemetry.CompactionDebtLimit/2 {
+			before := r.CompactionDebt()
+			if n, err := r.CompactParquet(ctx, compactor, 128); err != nil || n != 128 {
+				t.Fatalf("generation-0 retirement: %d %v", n, err)
+			}
+			if r.CompactionDebt() != before-128 {
+				t.Fatal("generation-0 debt did not retire")
+			}
+		}
+	}
+	fillDebt(telemetry.CompactionDebtLimit/2 - 1)
+	for _, want := range []int{8, 4} {
+		selected := selectCompactionBatchesWithHigherCap(r.Parquet.BatchMetadata(), 8,
+			compactionBatchCap(8, r.preemptedCompactionInputs), r.CompactionDebt())
+		if len(selected) != want || selected[0].Generation != 1 {
+			t.Fatalf("higher-generation retry: got %v, want %d generation-1 inputs", selected, want)
+		}
+		err := r.mergeCompaction(ctx, 1, len(selected), func(mergeCtx context.Context) error {
+			fillDebt(telemetry.CompactionDebtLimit)
+			<-mergeCtx.Done()
+			return mergeCtx.Err()
+		})
+		if !errors.Is(err, errCompactionPreempted) {
+			t.Fatalf("debt preemption: %v", err)
+		}
+		drainDebt()
+		if r.preemptedCompactionInputs != want || readCompactionAttempt(r.root) != 0 {
+			t.Fatal("generation-0 completion erased higher-generation retry size")
+		}
+	}
+	// Run the real native merge with the retained pair cap. Cross the threshold
+	// again before its atomic swap: the higher-generation output must publish
+	// even while ingest is blocked, then generation 0 must keep retiring debt.
+	pairPublisher := testParquetPublisherFunc(func(ctx context.Context, publish func(context.Context) error) error {
+		fillDebt(telemetry.CompactionDebtLimit)
+		return publish(ctx)
+	})
+	if n, err := r.CompactParquet(ctx, pairPublisher, 128); err != nil || n != 2 {
+		t.Fatalf("higher-generation publication: %d %v", n, err)
+	}
+	found := false
+	for _, batch := range r.Parquet.BatchMetadata() {
+		found = found || batch.Generation == 2 && batch.Logs == 2
+	}
+	if !found || r.preemptedCompactionInputs != 0 || r.CompactionDebt() != telemetry.CompactionDebtLimit {
+		t.Fatal("higher-generation success did not publish and reset its size")
+	}
+	drainDebt()
+	if n, err := r.CompactParquet(ctx, compactor, 8); err != nil || n != 8 {
+		t.Fatalf("restored higher-generation size: %d %v", n, err)
+	}
+	if r.RowCount() != uint64(sequence+1) {
+		t.Fatal("preemption or publication lost rows")
+	}
+}
diff --git a/internal/telemetry/store/compaction_pressure_test.go b/internal/telemetry/store/compaction_pressure_test.go
index d7dcdc6a..48b3177f 100644
--- a/internal/telemetry/store/compaction_pressure_test.go
+++ b/internal/telemetry/store/compaction_pressure_test.go
@@ -14,7 +14,7 @@ func TestPressuredCompactionBoundsRowsAndPreservesNormalCeilings(t *testing.T) {
 		if rows > telemetry.CompactionPressureMaxRows/2 {
 			small = append(small, genBatches(3, 0, 1151, 50_000)...)
 		}
-		selected := selectCompactionBatches(small, 128, telemetry.CompactionDebtLimit)
+		selected := selectCompactionBatchesWithHigherCap(small, 128, 128, telemetry.CompactionDebtLimit)
 		var total int64
 		for _, b := range selected {
 			total += compactionBatchRows(b)
@@ -28,13 +28,13 @@ func TestPressuredCompactionBoundsRowsAndPreservesNormalCeilings(t *testing.T) {
 		}
 	}
 	group := genBatches(2, 0, 128, 50_000)
-	if selected := selectCompactionBatches(group, 128, telemetry.CompactionDebtLimit/2-1); len(selected) != 128 {
+	if selected := selectCompactionBatchesWithHigherCap(group, 128, 128, telemetry.CompactionDebtLimit/2-1); len(selected) != 128 {
 		t.Fatalf("normal ceiling changed: %d", len(selected))
 	}
 	for i := range group {
 		group[i].Generation = 1
 	}
-	if selected := selectCompactionBatches(group, 128, telemetry.CompactionDebtLimit); len(selected) != 128 {
+	if selected := selectCompactionBatchesWithHigherCap(group, 128, 128, telemetry.CompactionDebtLimit); len(selected) != 128 {
 		t.Fatalf("higher generation ceiling changed: %d", len(selected))
 	}
 }
diff --git a/internal/telemetry/store/compaction_starvation_test.go b/internal/telemetry/store/compaction_starvation_test.go
index d770436b..3321999b 100644
--- a/internal/telemetry/store/compaction_starvation_test.go
+++ b/internal/telemetry/store/compaction_starvation_test.go
@@ -34,7 +34,7 @@ func TestSelectCompactionBatchesDoesNotStarveHigherGenerations(t *testing.T) {
 	// so both are valid candidates and the tie-break decides.
 	batches := append(genBatches(1, 0, 508, 1_000), genBatches(1, 2, 290, 200_000)...)
 
-	selected := selectCompactionBatches(batches, 128, 0)
+	selected := selectCompactionBatchesWithHigherCap(batches, 128, 128, 0)
 	if len(selected) == 0 {
 		t.Fatal("selected nothing with two compactable groups available")
 	}
@@ -50,7 +50,7 @@ func TestSelectCompactionBatchesDoesNotStarveHigherGenerations(t *testing.T) {
 func TestSelectCompactionBatchesStillPrefersTheOlderDay(t *testing.T) {
 	batches := append(genBatches(1, 0, 200, 1_000), genBatches(5, 2, 290, 200_000)...)
 
-	selected := selectCompactionBatches(batches, 128, 0)
+	selected := selectCompactionBatchesWithHigherCap(batches, 128, 128, 0)
 	if len(selected) == 0 {
 		t.Fatal("selected nothing")
 	}
@@ -69,7 +69,7 @@ func TestSelectCompactionBatchesPressureThreshold(t *testing.T) {
 			if debt >= telemetry.CompactionDebtLimit/2 {
 				want = 0
 			}
-			selected := selectCompactionBatches(batches, 128, debt)
+			selected := selectCompactionBatchesWithHigherCap(batches, 128, 128, debt)
 			if len(selected) < 2 || selected[0].Generation != want {
 				t.Fatalf("debt=%d selected=%v want generation %d", debt, selected, want)
 			}
@@ -85,12 +85,12 @@ func TestSelectCompactionBatchesPressureKeepsOldestEligibleGenerationZero(t *tes
 			batches = append(batches, genBatches(1, 0, 1, 1000)...)
 			batches = append(batches, genBatches(2, 0, 8, 1000)...)
 			batches = append(batches, genBatches(3, 0, 1200, 1000)...)
-			selected := selectCompactionBatches(batches, cap, telemetry.CompactionDebtLimit)
+			selected := selectCompactionBatchesWithHigherCap(batches, cap, cap, telemetry.CompactionDebtLimit)
 			if len(selected) != min(cap, 8) || selected[0].Generation != 0 || selected[0].MaxIngestedNanos/int64(24*time.Hour) != 2 {
 				t.Fatalf("selected=%v, want oldest eligible generation-0 group within cap %d", selected, cap)
 			}
 			// Clearing pressure restores the older higher-generation group.
-			selected = selectCompactionBatches(batches, cap, 0)
+			selected = selectCompactionBatchesWithHigherCap(batches, cap, cap, 0)
 			if len(selected) < 2 || selected[0].Generation != 2 || selected[0].MaxIngestedNanos/int64(24*time.Hour) != 1 {
 				t.Fatalf("normal order did not resume: %v", selected)
 			}
@@ -104,7 +104,7 @@ func TestSelectCompactionBatchesServesGenerationZeroWhenNothingIsPiledAbove(t *t
 	// Only generation 0 has enough to form a group.
 	batches := append(genBatches(1, 0, 200, 1_000), genBatches(1, 2, 3, 1_000)...)
 
-	selected := selectCompactionBatches(batches, 128, 0)
+	selected := selectCompactionBatchesWithHigherCap(batches, 128, 128, 0)
 	if len(selected) == 0 {
 		t.Fatal("selected nothing with a full generation-0 group available")
 	}
diff --git a/internal/telemetry/store/repository.go b/internal/telemetry/store/repository.go
index 06692273..f14643b4 100644
--- a/internal/telemetry/store/repository.go
+++ b/internal/telemetry/store/repository.go
@@ -35,6 +35,9 @@ type Repository struct {
 	lastPublished       int64
 	minimumStamp        int64
 	inFlight            map[int64]struct{}
+
+	// Guarded by compactionMu; generation-0 completions must not reset this.
+	preemptedCompactionInputs int
 }
 
 func Open(root string) (*Repository, error) {
diff --git a/internal/telemetry/store/repository_test.go b/internal/telemetry/store/repository_test.go
index b67b4c6b..4cbb2c73 100644
--- a/internal/telemetry/store/repository_test.go
+++ b/internal/telemetry/store/repository_test.go
@@ -494,10 +494,10 @@ func TestSelectCompactionBatchesRequiresFullSmallFileGroup(t *testing.T) {
 	for i := range batches {
 		batches[i] = telemetry.BatchMetadata{ID: fmt.Sprintf("batch-%d", i), MaxIngestedNanos: 1, Spans: 1}
 	}
-	if selected := selectCompactionBatches(batches[:groupSize-1], groupSize, 0); len(selected) != 0 {
+	if selected := selectCompactionBatchesWithHigherCap(batches[:groupSize-1], groupSize, groupSize, 0); len(selected) != 0 {
 		t.Fatalf("selected partial group of %d batches", len(selected))
 	}
-	if selected := selectCompactionBatches(batches, groupSize, 0); len(selected) != groupSize {
+	if selected := selectCompactionBatchesWithHigherCap(batches, groupSize, groupSize, 0); len(selected) != groupSize {
 		t.Fatalf("selected full group = %d, want %d", len(selected), groupSize)
 	}
 }
@@ -514,7 +514,7 @@ func TestSelectCompactionBatchesBuildsRowBoundedGroup(t *testing.T) {
 			Bytes: 1 << 20,
 		}
 	}
-	selected := selectCompactionBatches(batches, maxBatches, 0)
+	selected := selectCompactionBatchesWithHigherCap(batches, maxBatches, maxBatches, 0)
 	if len(selected) != 12 {
 		t.Fatalf("selected %d batches, want 12 below row limit", len(selected))
 	}
@@ -539,7 +539,7 @@ func TestSelectCompactionBatchesCombinesSaturatedLargeFiles(t *testing.T) {
 			Bytes: 1 << 20,
 		}
 	}
-	selected := selectCompactionBatches(batches, maxBatches, 0)
+	selected := selectCompactionBatchesWithHigherCap(batches, maxBatches, maxBatches, 0)
 	if len(selected) != len(batches) {
 		t.Fatalf("selected %d saturated large batches, want %d", len(selected), len(batches))
 	}
@@ -559,7 +559,7 @@ func TestSelectCompactionBatchesSkipsUncompactableOlderGroup(t *testing.T) {
 			Spans: 1,
 		}
 	}
-	selected := selectCompactionBatches(batches, minCompactionInputs, 0)
+	selected := selectCompactionBatchesWithHigherCap(batches, minCompactionInputs, minCompactionInputs, 0)
 	if len(selected) != minCompactionInputs || !strings.HasPrefix(selected[0].ID, "newer-") {
 		t.Fatalf("selection did not skip uncompactable older group: %#v", selected)
 	}