feat(observability): make a resumed or failed drive explain itself in the apply log

docs/apply-lifecycle.md

@@ -221,10 +221,13 @@ budget and window are per-apply — a new apply starts with a full budget.
 1. **Confirm the state.** Check the apply's status (CLI status command, or the
    summary comment on the PR). `failed_retryable` needs no action — recovery
    retries it automatically. Only permanent `failed` needs you.
-2. **Read the failure.** The apply's error message and the server logs (keyed
-   by the apply identifier) say what the engine rejected or what broke.
-   Fix the underlying cause if there is one — an invalid statement, a
-   permissions problem, an unreachable target.
+2. **Read the failure.** The apply's own log stream carries the cause, each
+   paused attempt with the recovery budget it had left, and the reason recovery
+   gave up — read it from the CLI (`schemabot logs <apply>`) or the PR summary
+   comment, no infrastructure access required. The server logs, keyed by the
+   apply identifier, are there when you need what happened around the apply
+   rather than to it. Fix the underlying cause if there is one — an invalid
+   statement, a permissions problem, an unreachable target.
 3. **Plan again.** Run a fresh plan against the same schema and environment
    (`schemabot plan`, or a new `schemabot plan` PR comment). The plan reflects
    whatever the failed apply already completed: tables that landed are not in

pkg/api/operator.go

@@ -238,6 +238,47 @@ func (s *Service) expireRetryableApplies(ctx context.Context, driverID int) {
 				"attempt", apply.Attempt,
 				"reason", expiration.Reason)...)
 		metrics.RecordOperatorResumeFailure(ctx, apply.Database, apply.Deployment, apply.Environment, string(expiration.Reason))
+		s.logApplyExpiration(ctx, apply, expiration.Reason)
+	}
+}
+
+// logApplyExpiration appends a durable apply log entry recording that operator
+// recovery gave up on the apply. Expiry is what makes a retryable failure
+// permanent, so without it the apply log ends on the last paused attempt and an
+// operator reading the CLI or the PR summary sees the apply reach a terminal
+// state with nothing stating why. Best-effort: a failed append must not stop
+// the driver from expiring the remaining applies.
+func (s *Service) logApplyExpiration(ctx context.Context, apply *storage.Apply, reason storage.RetryableExpirationReason) {
+	s.appendApplyLog(ctx, s.logger, &storage.ApplyLog{
+		ApplyID:   apply.ID,
+		Level:     storage.LogLevelError,
+		EventType: storage.LogEventError,
+		Source:    storage.LogSourceSchemaBot,
+		Message: fmt.Sprintf("Operator recovery gave up on the apply after %d of %d attempts (%s); it will not be retried automatically",
+			apply.Attempt, storage.MaxRecoveryAttempts, reason),
+		OldState:  state.Apply.FailedRetryable,
+		NewState:  state.Apply.Failed,
+		CreatedAt: s.clock.Now(),
+	}, "why recovery stopped", apply.LogAttrs()...)
+}
+
+// appendApplyLog writes one entry to the apply's own log stream, bounded so a
+// slow store cannot stall the driver. It is best-effort by contract: every
+// caller has already done the work the entry describes, and an entry that
+// cannot be written must not undo it. record names what an operator loses when
+// the entry does not land, so the warning says which part of the apply's
+// account is missing rather than only that a write failed.
+func (s *Service) appendApplyLog(ctx context.Context, logger *slog.Logger, entry *storage.ApplyLog, record string, logAttrs ...any) {
+	logStore := s.storage.ApplyLogs()
+	if logStore == nil {
+		logger.Warn("operator: no apply log store configured; the apply's own log will not state "+record, logAttrs...)
+		return
+	}
+	logCtx, cancel := context.WithTimeout(ctx, ApplyClaimLogTimeout)
+	defer cancel()
+	if err := logStore.Append(logCtx, entry); err != nil {
+		logger.Warn("operator: failed to append to the apply log; the apply's own log will not state "+record,
+			append(slices.Clone(logAttrs), "error", err)...)
 	}
 }
 
@@ -1416,14 +1457,7 @@ func (s *Service) resumeClaimedApplyWithOptions(ctx context.Context, driverID in
 // append must not block the resume, so the error is logged on the caller's
 // drive-scoped logger and the claim proceeds.
 func (s *Service) logApplyResumeClaim(ctx context.Context, logger *slog.Logger, driverID int, apply *storage.Apply) {
-	logStore := s.storage.ApplyLogs()
-	if logStore == nil {
-		logger.Warn("operator: no apply log store configured; apply claim will not appear in apply logs")
-		return
-	}
-	logCtx, cancel := context.WithTimeout(ctx, ApplyClaimLogTimeout)
-	defer cancel()
-	if err := logStore.Append(logCtx, &storage.ApplyLog{
+	s.appendApplyLog(ctx, logger, &storage.ApplyLog{
 		ApplyID:   apply.ID,
 		Level:     storage.LogLevelInfo,
 		EventType: storage.LogEventInfo,
@@ -1432,10 +1466,7 @@ func (s *Service) logApplyResumeClaim(ctx context.Context, logger *slog.Logger,
 		OldState:  apply.State,
 		NewState:  apply.State,
 		CreatedAt: s.clock.Now(),
-	}); err != nil {
-		logger.Warn("operator: failed to log apply claim; apply claim will not appear in apply logs",
-			"error", err)
-	}
+	}, "that a driver claimed it to resume it")
 }
 
 // failClaimedApplyAfterDrivePanic contains an engine-drive panic to the
@@ -1548,16 +1579,7 @@ func (s *Service) failClaimedApplyAfterDrivePanic(ctx context.Context, driverID
 // state without server logs. Best-effort: a failed append must not block
 // containment.
 func (s *Service) logApplyDrivePanicFailure(ctx context.Context, driverID int, apply *storage.Apply, previousState, errMsg string) {
-	logStore := s.storage.ApplyLogs()
-	if logStore == nil {
-		s.logger.Warn("operator: no apply log store configured; the drive panic will not appear in apply logs",
-			append(apply.LogAttrs(),
-				"driver", driverID)...)
-		return
-	}
-	logCtx, cancel := context.WithTimeout(ctx, ApplyClaimLogTimeout)
-	defer cancel()
-	if err := logStore.Append(logCtx, &storage.ApplyLog{
+	s.appendApplyLog(ctx, s.logger, &storage.ApplyLog{
 		ApplyID:   apply.ID,
 		Level:     storage.LogLevelError,
 		EventType: storage.LogEventError,
@@ -1566,12 +1588,7 @@ func (s *Service) logApplyDrivePanicFailure(ctx context.Context, driverID int, a
 		OldState:  previousState,
 		NewState:  apply.State,
 		CreatedAt: s.clock.Now(),
-	}); err != nil {
-		s.logger.Warn("operator: failed to log drive panic failure; the failure will not appear in apply logs",
-			append(apply.LogAttrs(),
-				"driver", driverID,
-				"error", err)...)
-	}
+	}, "that a contained drive panic failed it", append(apply.LogAttrs(), "driver", driverID)...)
 }
 
 // startApplyOperationHeartbeat refreshes the claimed operation row's lease while

pkg/api/operator_test.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"log/slog"
 	"os"
 	"sync"
@@ -221,7 +222,7 @@ func TestResumeClaimedApply_DriveLogsCarryApplyIdentity(t *testing.T) {
 
 	// The call site passes only the message — every identity attr must come
 	// from the bound logger.
-	noStore := requireLogLine(t, lines, "operator: no apply log store configured; apply claim will not appear in apply logs")
+	noStore := requireLogLine(t, lines, "operator: no apply log store configured; the apply's own log will not state that a driver claimed it to resume it")
 	assertDriveIdentity(noStore)
 
 	resumedLine := requireLogLine(t, lines, "operator: resumed apply")
@@ -242,6 +243,40 @@ func (s *expiringApplyStore) ExpireRetryable(context.Context) ([]*storage.Retrya
 	return s.expirations, nil
 }
 
+// Expiry is what makes a retryable failure permanent, so it belongs in the
+// apply's own log stream: that stream is what the CLI and the PR summary
+// render, and an apply whose last entry is a paused attempt reads as one that
+// went terminal for no stated reason.
+func TestExpireRetryableApplies_RecordsWhyRecoveryStoppedInTheApplyLog(t *testing.T) {
+	apply := &storage.Apply{
+		ID:              42,
+		ApplyIdentifier: "apply-42",
+		Database:        "appdb",
+		Environment:     "staging",
+		State:           state.Apply.Failed,
+		Attempt:         storage.MaxRecoveryAttempts,
+	}
+	applyLogs := &capturingApplyLogStore{}
+	svc := New(&mockStorageWithApplyStores{
+		applies: &expiringApplyStore{expirations: []*storage.RetryableApplyExpiration{
+			{Apply: apply, Reason: storage.RetryableExpirationAttemptBudget},
+		}},
+		applyLogs: applyLogs,
+	}, testServerConfig(), nil, slog.Default())
+
+	svc.expireRetryableApplies(t.Context(), 1)
+
+	require.Len(t, applyLogs.logs, 1)
+	entry := applyLogs.logs[0]
+	assert.Equal(t, storage.LogLevelError, entry.Level)
+	assert.Equal(t, int64(42), entry.ApplyID)
+	assert.Contains(t, entry.Message,
+		fmt.Sprintf("%d of %d attempts", storage.MaxRecoveryAttempts, storage.MaxRecoveryAttempts))
+	assert.Contains(t, entry.Message, string(storage.RetryableExpirationAttemptBudget))
+	assert.Equal(t, state.Apply.FailedRetryable, entry.OldState)
+	assert.Equal(t, state.Apply.Failed, entry.NewState)
+}
+
 // A retryable-apply expiry is a control-plane lifecycle transition an operator
 // triages from logs alone, so the expiry line must carry the apply's full
 // triage attributes — including external_id, the join key to the data plane's

pkg/tern/apply_failure_log_test.go

@@ -0,0 +1,175 @@
+package tern
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/block/schemabot/pkg/state"
+	"github.com/block/schemabot/pkg/storage"
+)
+
+// capturingApplyLogStore records every entry written to an apply's log stream.
+type capturingApplyLogStore struct {
+	storage.ApplyLogStore
+	entries []*storage.ApplyLog
+}
+
+func (s *capturingApplyLogStore) Append(_ context.Context, entry *storage.ApplyLog) error {
+	s.entries = append(s.entries, entry)
+	return nil
+}
+
+func newFailureLogTestClient(apply *storage.Apply, tasks []*storage.Task) (*LocalClient, *capturingApplyLogStore) {
+	logs := &capturingApplyLogStore{}
+	return &LocalClient{
+		config: LocalConfig{Database: "testdb", Type: storage.DatabaseTypeMySQL},
+		storage: &controlTestStorage{
+			applies:         &controlTestApplyStore{apply: apply},
+			tasks:           &controlTestTaskStore{tasks: tasks},
+			applyLogs:       logs,
+			controlRequests: &testControlRequestStore{},
+		},
+		logger: slog.Default(),
+	}, logs
+}
+
+func failureLogTestApply(applyState string, attempt int) *storage.Apply {
+	return &storage.Apply{
+		ID:              7,
+		ApplyIdentifier: "apply-7",
+		Database:        "orders",
+		Environment:     "staging",
+		State:           applyState,
+		Attempt:         attempt,
+	}
+}
+
+// The apply log is the only failure record an operator reads from the CLI or
+// the PR summary comment, so an apply that fails must say so there — not only
+// in the server logs, where a reader of the apply's own history would see it
+// reach a terminal state with nothing stating why.
+func TestFailApplyWithTasksRecordsTheFailureInTheApplyLog(t *testing.T) {
+	apply := failureLogTestApply(state.Apply.Running, 0)
+	task := &storage.Task{ID: 1, TaskIdentifier: "task-1", ApplyID: apply.ID, TableName: "orders", State: state.Task.Running}
+	client, logs := newFailureLogTestClient(apply, []*storage.Task{task})
+
+	client.failApplyWithTasks(t.Context(), apply, []*storage.Task{task}, "engine lost its connection to the target")
+
+	require.Len(t, logs.entries, 1)
+	entry := logs.entries[0]
+	assert.Equal(t, storage.LogLevelError, entry.Level)
+	assert.Equal(t, storage.LogEventError, entry.EventType)
+	assert.Equal(t, "Apply failed: engine lost its connection to the target", entry.Message)
+	assert.Equal(t, state.Apply.Running, entry.OldState)
+	assert.Equal(t, state.Apply.Failed, entry.NewState)
+}
+
+// A retryable failure is the state operator recovery keeps re-driving, so each
+// paused attempt records the budget it has left. Without it the apply log shows
+// only the gaps between attempts, and the retry budget drains invisibly. The
+// countdown is derived from the apply's own attempt counter, so it can never
+// run past the budget it is measured against.
+func TestMarkApplyRetryableWithTasksRecordsTheRemainingBudget(t *testing.T) {
+	apply := failureLogTestApply(state.Apply.Running, 3)
+	task := &storage.Task{ID: 1, TaskIdentifier: "task-1", ApplyID: apply.ID, TableName: "orders", State: state.Task.Running}
+	client, logs := newFailureLogTestClient(apply, []*storage.Task{task})
+
+	client.markApplyRetryableWithTasks(t.Context(), apply, []*storage.Task{task}, "target refused the connection")
+
+	require.Len(t, logs.entries, 1)
+	entry := logs.entries[0]
+	assert.Equal(t, storage.LogLevelWarn, entry.Level, "a paused attempt is not yet a permanent failure")
+	assert.Contains(t, entry.Message,
+		fmt.Sprintf("%d of %d recovery attempts remaining", storage.MaxRecoveryAttempts-3, storage.MaxRecoveryAttempts))
+	assert.Contains(t, entry.Message, "target refused the connection")
+	assert.Equal(t, state.Apply.Running, entry.OldState)
+	assert.Equal(t, state.Apply.FailedRetryable, entry.NewState)
+}
+
+// The last drive operator recovery is allowed to make reports an exhausted
+// budget, so an operator reading the apply log sees the retry budget reach its
+// limit rather than a figure that runs past it.
+func TestMarkApplyRetryableWithTasksReportsAnExhaustedBudget(t *testing.T) {
+	apply := failureLogTestApply(state.Apply.Running, storage.MaxRecoveryAttempts)
+	client, logs := newFailureLogTestClient(apply, nil)
+
+	client.markApplyRetryableWithTasks(t.Context(), apply, nil, "target refused the connection")
+
+	require.Len(t, logs.entries, 1)
+	assert.Contains(t, logs.entries[0].Message,
+		fmt.Sprintf("0 of %d recovery attempts remaining", storage.MaxRecoveryAttempts))
+}
+
+// An apply that another driver already settled is not this drive's to fail: the
+// stored verdict stands, and writing a second failure record would report a
+// state transition that never happened.
+func TestFailApplyWithTasksLeavesASettledApplyUnrecorded(t *testing.T) {
+	apply := failureLogTestApply(state.Apply.Cancelled, 0)
+	client, logs := newFailureLogTestClient(apply, nil)
+
+	client.failApplyWithTasks(t.Context(), apply, nil, "engine lost its connection to the target")
+
+	assert.Empty(t, logs.entries)
+	assert.Equal(t, state.Apply.Cancelled, apply.State)
+}
+
+// The default sequential MySQL path is the one most applies take, so a table
+// that fails there reaches the operator through the apply's own log stream like
+// every other path — with the cause named, and with the budget countdown when
+// the failure is one recovery will re-drive.
+func TestFinalizeSequentialApplyRecordsAPermanentFailure(t *testing.T) {
+	apply := failureLogTestApply(state.Apply.Running, 0)
+	failed := &storage.Task{
+		ID: 1, TaskIdentifier: "task-1", ApplyID: apply.ID, TableName: "orders",
+		State: state.Task.Failed, ErrorMessage: "engine refused the statement",
+	}
+	client, logs := newFailureLogTestClient(apply, []*storage.Task{failed})
+
+	client.finalizeSequentialApply(t.Context(), apply, []*storage.Task{failed}, failed, false)
+
+	require.Len(t, logs.entries, 1)
+	entry := logs.entries[0]
+	assert.Equal(t, storage.LogLevelError, entry.Level)
+	assert.Contains(t, entry.Message, "Apply failed:")
+	assert.Contains(t, entry.Message, "orders")
+	assert.Contains(t, entry.Message, "engine refused the statement")
+	assert.Equal(t, state.Apply.Running, entry.OldState)
+	assert.Equal(t, state.Apply.Failed, entry.NewState)
+}
+
+func TestFinalizeSequentialApplyRecordsARetryablePause(t *testing.T) {
+	apply := failureLogTestApply(state.Apply.Running, 3)
+	failed := &storage.Task{
+		ID: 1, TaskIdentifier: "task-1", ApplyID: apply.ID, TableName: "orders",
+		State: state.Task.FailedRetryable, ErrorMessage: "target refused the connection",
+	}
+	client, logs := newFailureLogTestClient(apply, []*storage.Task{failed})
+
+	client.finalizeSequentialApply(t.Context(), apply, []*storage.Task{failed}, failed, false)
+
+	require.Len(t, logs.entries, 1)
+	entry := logs.entries[0]
+	assert.Equal(t, storage.LogLevelWarn, entry.Level)
+	assert.Contains(t, entry.Message,
+		fmt.Sprintf("%d of %d recovery attempts remaining", storage.MaxRecoveryAttempts-3, storage.MaxRecoveryAttempts))
+	assert.Contains(t, entry.Message, "target refused the connection")
+	assert.Equal(t, state.Apply.FailedRetryable, entry.NewState)
+}
+
+// A sequential apply that succeeds or that an operator stopped is not a failure,
+// and recording one would report a cause that does not exist.
+func TestFinalizeSequentialApplyRecordsNothingWithoutAFailure(t *testing.T) {
+	apply := failureLogTestApply(state.Apply.Running, 0)
+	done := &storage.Task{ID: 1, TaskIdentifier: "task-1", ApplyID: apply.ID, TableName: "orders", State: state.Task.Completed}
+	client, logs := newFailureLogTestClient(apply, []*storage.Task{done})
+
+	client.finalizeSequentialApply(t.Context(), apply, []*storage.Task{done}, nil, false)
+
+	assert.Empty(t, logs.entries)
+	assert.Equal(t, state.Apply.Completed, apply.State)
+}

pkg/tern/grpc_client_test.go

@@ -820,7 +820,12 @@ type mockApplyLogStore struct {
 	recentLimit int
 }
 
-func (m *mockApplyLogStore) Append(_ context.Context, log *storage.ApplyLog) error {
+// Append records the line the way the real store does: a cancelled context is
+// a failed write, not a silently dropped one.
+func (m *mockApplyLogStore) Append(ctx context.Context, log *storage.ApplyLog) error {
+	if err := ctx.Err(); err != nil {
+		return fmt.Errorf("append apply log for apply %d: %w", log.ApplyID, err)
+	}
 	stored := *log
 	m.logs = append(m.logs, &stored)
 	return nil