fix(tern): sequential operation-scoped drives leave the parent apply to the projection

pkg/tern/control_requests.go

@@ -3,6 +3,7 @@ package tern
 import (
 	"context"
 	"fmt"
+	"log/slog"
 	"time"
 
 	"github.com/block/schemabot/pkg/state"
@@ -102,7 +103,7 @@ func failPendingControlRequests(ctx context.Context, store storage.Storage, appl
 	return nil
 }
 
-func markApplyCuttingOverForControlRequest(ctx context.Context, store storage.Storage, apply *storage.Apply) error {
+func markApplyCuttingOverForControlRequest(ctx context.Context, store storage.Storage, apply *storage.Apply, logger *slog.Logger) error {
 	if !state.IsState(apply.State, state.Apply.WaitingForCutover) && !state.IsRunningApplyState(apply.State) {
 		return nil
 	}
@@ -117,6 +118,16 @@ func markApplyCuttingOverForControlRequest(ctx context.Context, store storage.St
 	now := time.Now()
 	apply.State = state.Apply.CuttingOver
 	apply.UpdatedAt = now
+	// A multi-operation drive owns only its operation: the parent cutting_over
+	// write is the operator's projection to make — a direct write here fails
+	// closed under the operation-only lease and would block a cutover the
+	// engine is ready to accept. The in-memory transition still stands so this
+	// drive proceeds to dispatch the cutover.
+	if suppressParentApplyWrites(ctx) {
+		logger.Info("pending cutover request accepted under operation lease; parent cutting_over state is the operator's projection",
+			"state", apply.State)
+		return nil
+	}
 	if err := applyStore.Update(ctx, apply); err != nil {
 		*apply = previous
 		return fmt.Errorf("mark apply %s cutting over for pending cutover request: %w", apply.ApplyIdentifier, err)

pkg/tern/grpc_client.go

@@ -622,7 +622,7 @@ func (c *GRPCClient) processPendingCutoverControlRequest(ctx context.Context, ap
 		message := "schema change has a pending stop request; cutover is blocked until stop is processed"
 		return fmt.Errorf("process pending gRPC cutover for apply %s: %s", apply.ApplyIdentifier, message)
 	}
-	if err := markApplyCuttingOverForControlRequest(ctx, c.storage, apply); err != nil {
+	if err := markApplyCuttingOverForControlRequest(ctx, c.storage, apply, logger); err != nil {
 		return err
 	}
 	resp, err := c.client.Cutover(ctx, &ternv1.CutoverRequest{

pkg/tern/local_apply_failure.go

@@ -51,6 +51,15 @@ func (c *LocalClient) failApplyWithTasks(ctx context.Context, apply *storage.App
 	}
 
 	logger := c.logger.With(apply.IdentityLogAttrs()...)
+	// A multi-operation drive owns only its operation: the failed tasks above
+	// carry the outcome, the operator derives the operation row from them and
+	// projects the parent, so the parent failed write, apply-level metric, and
+	// failure log are the operator's to make.
+	if suppressParentApplyWrites(ctx) {
+		logger.Info("operation drive failed its tasks; operator derives the operation row and projects the parent",
+			"error_message", errMsg)
+		return
+	}
 	// Re-read the apply from storage — Stop() may have already set a terminal
 	// state (e.g., cancelled) between when the engine error occurred and now.
 	fresh, err := c.storage.Applies().Get(ctx, apply.ID)
@@ -90,6 +99,15 @@ func (c *LocalClient) markApplyRetryableWithTasks(ctx context.Context, apply *st
 	}
 
 	logger := c.logger.With(apply.IdentityLogAttrs()...)
+	// A multi-operation drive owns only its operation: the failed_retryable
+	// tasks above carry the outcome, the operator derives the operation row from
+	// them and projects the parent, so the parent retryable write, apply-level
+	// metric, retry log, and observer are the operator's to make.
+	if suppressParentApplyWrites(ctx) {
+		logger.Info("operation drive paused its tasks for retry; operator derives the operation row and projects the parent",
+			"error_message", errMsg)
+		return
+	}
 	// Re-read the apply from storage; Stop() may have already moved it to a
 	// terminal state between the engine error and this update.
 	fresh, err := c.storage.Applies().Get(ctx, apply.ID)

pkg/tern/local_apply_sequential.go

@@ -643,6 +643,26 @@ func (c *LocalClient) shouldRetryEngineError(err error) bool {
 func (c *LocalClient) finalizeSequentialApply(ctx context.Context, apply *storage.Apply, tasks []*storage.Task, failedTask *storage.Task, stoppedByUser bool) {
 	now := time.Now()
 	logger := c.logger.With(apply.IdentityLogAttrs()...)
+	// A multi-operation drive owns only its operation: the tasks it drove carry
+	// the outcome, the operator derives the operation row from them and projects
+	// the parent, so the parent terminal write, control-request completion,
+	// apply-level metric, and terminal observer are all the operator's to make.
+	// Pending tasks after a failed one are still this drive's to settle, and the
+	// in-memory outcome is still adopted so the drive's own logs report what the
+	// operation settled to rather than the projection's stale running state.
+	if suppressParentApplyWrites(ctx) {
+		if failedTask != nil && failedTask.State != state.Task.FailedRetryable {
+			for _, task := range tasks {
+				if task.State == state.Task.Pending {
+					c.transitionTaskState(ctx, task, 0, state.Task.Cancelled, "")
+				}
+			}
+		}
+		adoptSequentialOutcome(apply, failedTask, stoppedByUser, now)
+		logger.Info("sequential operation drive settled; operator derives the operation row and projects the parent",
+			"stopped_by_user", stoppedByUser, "failed_task", failedTask != nil, "settled_state", apply.State)
+		return
+	}
 	if freshApply, err := c.storage.Applies().Get(ctx, apply.ID); err != nil {
 		logger.Error("failed to reload apply before sequential finalization",
 			append(apply.MutableLogAttrs(), "error", err)...)
@@ -658,27 +678,14 @@ func (c *LocalClient) finalizeSequentialApply(ctx context.Context, apply *storag
 		return
 	}
 	previousState := apply.State
-	switch {
-	case failedTask != nil && failedTask.State == state.Task.FailedRetryable:
-		apply.State = state.Apply.FailedRetryable
-		apply.ErrorMessage = fmt.Sprintf("table %s failed: %s", failedTask.TableName, failedTask.ErrorMessage)
-		apply.CompletedAt = nil
-	case failedTask != nil:
-		apply.State = state.Apply.Failed
-		apply.ErrorMessage = fmt.Sprintf("table %s failed: %s", failedTask.TableName, failedTask.ErrorMessage)
-		apply.CompletedAt = &now
+	if failedTask != nil && failedTask.State != state.Task.FailedRetryable {
 		for _, task := range tasks {
 			if task.State == state.Task.Pending {
 				c.transitionTaskState(ctx, task, 0, state.Task.Cancelled, "")
 			}
 		}
-	case stoppedByUser:
-		apply.State = state.Apply.Stopped
-	default:
-		apply.State = state.Apply.Completed
-		apply.CompletedAt = &now
 	}
-	apply.UpdatedAt = now
+	adoptSequentialOutcome(apply, failedTask, stoppedByUser, now)
 	if err := c.storage.Applies().Update(ctx, apply); err != nil {
 		logger.Error("failed to update apply state", append(apply.MutableLogAttrs(), "error", err)...)
 	} else {
@@ -714,3 +721,26 @@ func (c *LocalClient) finalizeSequentialApply(ctx context.Context, apply *storag
 		c.clearObserver(apply.ID)
 	}
 }
+
+// adoptSequentialOutcome mutates the in-memory apply to the outcome its
+// sequential task results settle to: failed_retryable for a retryable task
+// failure, failed for a permanent one, stopped for an operator stop, completed
+// otherwise. Callers own persisting (or not persisting) the mutated row.
+func adoptSequentialOutcome(apply *storage.Apply, failedTask *storage.Task, stoppedByUser bool, now time.Time) {
+	switch {
+	case failedTask != nil && failedTask.State == state.Task.FailedRetryable:
+		apply.State = state.Apply.FailedRetryable
+		apply.ErrorMessage = fmt.Sprintf("table %s failed: %s", failedTask.TableName, failedTask.ErrorMessage)
+		apply.CompletedAt = nil
+	case failedTask != nil:
+		apply.State = state.Apply.Failed
+		apply.ErrorMessage = fmt.Sprintf("table %s failed: %s", failedTask.TableName, failedTask.ErrorMessage)
+		apply.CompletedAt = &now
+	case stoppedByUser:
+		apply.State = state.Apply.Stopped
+	default:
+		apply.State = state.Apply.Completed
+		apply.CompletedAt = &now
+	}
+	apply.UpdatedAt = now
+}

pkg/tern/local_control.go

@@ -288,7 +288,7 @@ func (c *LocalClient) processPendingCutoverControlRequest(ctx context.Context, a
 		message := "schema change has a pending stop request; cutover is blocked until stop is processed"
 		return fmt.Errorf("process pending cutover for apply %s: %s", apply.ApplyIdentifier, message)
 	}
-	if err := markApplyCuttingOverForControlRequest(ctx, c.storage, apply); err != nil {
+	if err := markApplyCuttingOverForControlRequest(ctx, c.storage, apply, logger); err != nil {
 		return err
 	}
 	resp, err := c.cutover(ctx, &ternv1.CutoverRequest{
@@ -1653,6 +1653,25 @@ func (c *LocalClient) settleControlForCompletedEngineChange(ctx context.Context,
 	if applyID == 0 {
 		return 0, fmt.Errorf("%s found the schema change already completed on the engine, but resolved no apply to settle: %w", operation, rejection)
 	}
+	// A multi-operation drive owns only its operation: the tasks settled above
+	// carry the completed outcome, the operator derives the operation row from
+	// them and projects the parent completed, so the parent write, apply event,
+	// and terminal observer are the operator's to make — a direct write here
+	// fails closed under the operation-only lease and would turn an accepted
+	// settle into a drive error the claim loop re-runs forever.
+	if suppressParentApplyWrites(ctx) {
+		attrs := []any{"database", c.config.Database}
+		if targetApply != nil {
+			attrs = targetApply.LogAttrs()
+		}
+		c.logger.Info("engine rejected "+operation+" because the schema change already completed; operation drive settled its tasks and the operator projects the parent",
+			append(attrs,
+				"requested_by", caller,
+				"completed_task_count", completedCount,
+				"terminal_task_count", skippedCount,
+				"error", rejection)...)
+		return skippedCount, nil
+	}
 	apply, err := c.storage.Applies().Get(ctx, applyID)
 	if err != nil {
 		return 0, fmt.Errorf("load apply %d to settle %s for a completed schema change: %w", applyID, operation, err)