diff --git a/pkg/viewlog/ingest.go b/pkg/viewlog/ingest.go
index 3749836e7..2538213ac 100644
--- a/pkg/viewlog/ingest.go
+++ b/pkg/viewlog/ingest.go
@@ -138,9 +138,26 @@ func RunIngest(ctx context.Context, in IngestInput) (*IngestResult, error) {
 			}
 			continue
 		}
+		// Request re-aggregation BEFORE marking the part complete. A
+		// completed part is never revisited, so an enqueue that fails
+		// after completion would leave that window's published count
+		// permanently missing the CDN's segments. Failing here instead
+		// releases the claim; the next pass re-ingests the part, which
+		// is idempotent (output keys derive from part ID + window, so
+		// the rewrite lands on the same files with the same content).
+		if err := requestReaggregation(ctx, in, part.ID, pr.windows); err != nil {
+			log.Error(ctx, "viewlog: request re-aggregation failed", "source", in.SourceName, "part", part.ID, "error", err)
+			if rerr := in.Cursor.ReleasePart(ctx, in.SourceName, part.ID); rerr != nil {
+				log.Error(ctx, "viewlog: release part claim", "part", part.ID, "error", rerr)
+			}
+			if firstErr == nil {
+				firstErr = err
+			}
+			continue
+		}
 		if err := in.Cursor.CompletePart(ctx, in.SourceName, part.ID, pr.events); err != nil {
-			// The files are visible; a re-run would duplicate them.
-			// Surface loudly rather than release.
+			// Files + re-aggregation requests are in place; the next
+			// pass will redo both harmlessly. Surface it anyway.
 			return res, fmt.Errorf("viewlog: complete part %s: %w", part.ID, err)
 		}
 		res.PartsIngested++
@@ -150,20 +167,26 @@ func RunIngest(ctx context.Context, in IngestInput) (*IngestResult, error) {
 		log.Log(ctx, "viewlog: ingested cdn log part",
 			"source", in.SourceName, "part", part.ID, "size", part.Size,
 			"events", pr.events, "skipped", pr.skipped, "windows", len(pr.windows))
-		if in.Reaggregate != nil {
-			for _, w := range pr.windows {
-				if err := in.Reaggregate(ctx, part.ID, w, w.Add(in.Window)); err != nil {
-					log.Error(ctx, "viewlog: request re-aggregation", "part", part.ID, "window", w, "error", err)
-					if firstErr == nil {
-						firstErr = err
-					}
-				}
-			}
-		}
 	}
 	return res, firstErr
 }
 
+// requestReaggregation asks for every touched window to be recomputed.
+// Stops at the first failure: the caller releases the part and the
+// whole thing is retried next pass (duplicate requests dedup on their
+// task key).
+func requestReaggregation(ctx context.Context, in IngestInput, partID string, windows []time.Time) error {
+	if in.Reaggregate == nil {
+		return nil
+	}
+	for _, w := range windows {
+		if err := in.Reaggregate(ctx, partID, w, w.Add(in.Window)); err != nil {
+			return fmt.Errorf("window %s: %w", w.Format(time.RFC3339), err)
+		}
+	}
+	return nil
+}
+
 type partResult struct {
 	events, skipped int
 	windows         []time.Time
diff --git a/pkg/viewlog/ingest_test.go b/pkg/viewlog/ingest_test.go
index 8eca8b90d..83570b985 100644
--- a/pkg/viewlog/ingest_test.go
+++ b/pkg/viewlog/ingest_test.go
@@ -254,6 +254,50 @@ func TestRunIngestFailedPartIsReleasedAndRetried(t *testing.T) {
 	require.Len(t, keys, 2)
 }
 
+// TestRunIngestReaggregateFailureRetriesPart: if requesting
+// re-aggregation fails, the part must not be marked complete, or the
+// window's published count would never pick up the CDN's segments.
+// The next pass redoes the part; the rewrite lands on the same keys.
+func TestRunIngestReaggregateFailureRetriesPart(t *testing.T) {
+	root := t.TempDir()
+	store, err := blob.NewFileStore(root)
+	require.NoError(t, err)
+	day := time.Date(2026, 9, 5, 0, 0, 0, 0, time.UTC)
+	src := &fakeSource{
+		parts:    []cdn.Part{{ID: "p/a.gzip", Day: day}},
+		requests: map[string][]cdn.Request{"p/a.gzip": {segReq(day.Add(time.Hour), 200, 10, "bafyblob", "tid1")}},
+	}
+	cursor := newMemCursor()
+	failing := true
+	var reaggs int
+	in := IngestInput{
+		Store: store, Source: src, SourceName: "bunny", Salts: NewSaltManager(newMemSaltStorage()),
+		Cursor: cursor, Window: 5 * time.Minute,
+		Reaggregate: func(ctx context.Context, part string, start, end time.Time) error {
+			reaggs++
+			if failing {
+				return errors.New("queue down")
+			}
+			return nil
+		},
+		Now: func() time.Time { return day.Add(26 * time.Hour) },
+	}
+	res, err := RunIngest(context.Background(), in)
+	require.Error(t, err)
+	require.Equal(t, 0, res.PartsIngested)
+	require.Empty(t, cursor.completed, "part is not complete after a failed re-aggregation request")
+
+	failing = false
+	res, err = RunIngest(context.Background(), in)
+	require.NoError(t, err)
+	require.Equal(t, 1, res.PartsIngested)
+	require.Equal(t, 2, reaggs)
+	require.Equal(t, 2, src.reads, "the part was re-read on retry")
+	keys, err := store.List(context.Background(), viewLogsPrefix)
+	require.NoError(t, err)
+	require.Len(t, keys, 1, "the retry overwrote the same window file rather than adding a duplicate")
+}
+
 func TestIngestTaskKeys(t *testing.T) {
 	start := time.Date(2026, 9, 5, 10, 0, 0, 0, time.UTC)
 	end := start.Add(5 * time.Minute)