fix(tern): settle sequential tasks when the engine loses in-flight work

pkg/engine/engine.go

@@ -154,6 +154,40 @@ func ProgressIsExternallyAuthoritative(eng Engine) bool {
 	return ok && auth.ProgressIsExternallyAuthoritative()
 }
 
+// SynchronousWorkRegistration is an optional interface for engines whose Apply
+// registers accepted work before it returns, so the engine can never report
+// pending for work it has accepted but has not begun executing.
+//
+// This distinction decides how a driver reads a pending progress report for a
+// task whose durable state says the work is in flight. Pending is an overloaded
+// report. An engine that provisions resources after accepting the work — cutting
+// a branch, opening and validating a deploy request — reports pending for real,
+// healthy work for as long as that setup takes, so a driver has to give it time
+// before concluding anything from the report. An engine that registers the work
+// synchronously has no such phase: once Apply has returned, the work is either
+// running or it is gone, and a single pending report is already conclusive.
+//
+// Engines that do not implement this interface are treated as having a setup
+// phase. That is the safe default for a healthy schema change: an undeclared
+// engine is given the driver's full trust budget, so provisioning is never
+// mistaken for lost work and a change that was about to run is never bounced.
+type SynchronousWorkRegistration interface {
+	// RegistersWorkSynchronously reports whether Apply registers accepted work
+	// before returning, which makes a pending progress report conclusive
+	// evidence that accepted work is gone rather than not yet started.
+	RegistersWorkSynchronously() bool
+}
+
+// RegistersWorkSynchronously reports whether eng declares that Apply registers
+// accepted work before returning. Engines that do not implement
+// SynchronousWorkRegistration are treated as having a post-acceptance setup
+// phase, so a pending report about in-flight work is never read as conclusive
+// on its own.
+func RegistersWorkSynchronously(eng Engine) bool {
+	reg, ok := eng.(SynchronousWorkRegistration)
+	return ok && reg.RegistersWorkSynchronously()
+}
+
 // DeferredCutoverSignalRequest identifies the target database whose deferred
 // cutover signal should be inspected.
 type DeferredCutoverSignalRequest struct {

pkg/engine/postgres/postgres.go

@@ -360,6 +360,18 @@ func sortedKeys[V any](values map[string]V) []string {
 	return keys
 }
 
+// RegistersWorkSynchronously reports that Apply records the accepted schema
+// change on this engine before it returns, so there is no window in which the
+// engine has accepted work it cannot yet describe. The statement executes in a
+// goroutine of this process with nothing to provision first, and the tracked
+// progress is claimed under the engine mutex before Apply returns; Drain is
+// the only writer that clears it, and a drained engine's work is not coming
+// back. A pending progress report for a task a driver believes is in flight is
+// therefore conclusive rather than a phase to wait out.
+func (e *Engine) RegistersWorkSynchronously() bool {
+	return true
+}
+
 // Drain blocks until every background apply goroutine has finished, then
 // clears the tracked schema change so the next Progress reports the idle
 // sentinel. Resume and recovery paths call this before re-planning so a

pkg/engine/postgres/postgres_test.go

@@ -356,6 +356,19 @@ func TestLifecycleControlsDeclineAsUnsupported(t *testing.T) {
 	}
 }
 
+// The PostgreSQL engine runs each statement in a goroutine of this process and
+// claims the tracked progress before Apply returns, so it declares its work
+// registration synchronous. A driver reads that declaration to decide whether
+// a pending progress report about work it believes is in flight is conclusive.
+func TestRegistersWorkSynchronously(t *testing.T) {
+	eng := New()
+
+	assert.True(t, eng.RegistersWorkSynchronously(),
+		"the engine claims the tracked schema change before Apply returns")
+	assert.True(t, engine.RegistersWorkSynchronously(eng),
+		"the package helper resolves the engine's declaration")
+}
+
 // A zero ceiling means unset and adopts the default, so a zero-valued client
 // config preserves the stock ceiling instead of disabling the size guard.
 func TestNewWithTableSizeLimitTreatsZeroAsUnset(t *testing.T) {

pkg/engine/spirit/spirit.go

@@ -288,6 +288,18 @@ func (e *Engine) DebugLogs() bool {
 	return e.debugLogs.Load()
 }
 
+// RegistersWorkSynchronously reports that Apply records the accepted schema
+// change on this engine before it returns, so there is no window in which Spirit
+// has accepted work it cannot yet describe. Spirit executes in a goroutine of
+// this process with nothing to provision first, and the tracked state is
+// published under the engine mutex before Apply returns; Drain and the cancel
+// path are the only writers that clear it, and both mean the work is not coming
+// back. A pending progress report for a task a driver believes is in flight is
+// therefore conclusive rather than a phase to wait out.
+func (e *Engine) RegistersWorkSynchronously() bool {
+	return true
+}
+
 // Drain waits for any in-flight migration goroutine to complete and clears the
 // running migration state. This ensures DB connections from a previous run are
 // fully released before new operations begin.

pkg/engine/spirit/spirit_test.go

@@ -318,3 +318,16 @@ func TestBuildSpiritTableProgress(t *testing.T) {
 		})
 	}
 }
+
+// Spirit runs the schema change in a goroutine of this process and publishes
+// the tracked state before Apply returns, so it declares its work registration
+// synchronous. A driver reads that declaration to decide whether a pending
+// progress report about work it believes is in flight is conclusive.
+func TestRegistersWorkSynchronously(t *testing.T) {
+	eng := New(Config{})
+
+	assert.True(t, eng.RegistersWorkSynchronously(),
+		"Spirit publishes the tracked schema change before Apply returns")
+	assert.True(t, engine.RegistersWorkSynchronously(eng),
+		"the package helper resolves Spirit's declaration")
+}

pkg/tern/local_apply_grouped.go

@@ -568,7 +568,7 @@ func (c *LocalClient) deriveAggregateApplyState(ctx context.Context, apply *stor
 // pollForCompletionAtomic polls the engine for progress in atomic mode (all tasks share state).
 func (c *LocalClient) pollForCompletionAtomic(ctx context.Context, apply *storage.Apply, tasks []*storage.Task, creds *engine.Credentials, resumeState *engine.ResumeState, options map[string]string, releaseAtCutoverBarrier bool) {
 	eng := c.getEngine()
-	ticker := time.NewTicker(500 * time.Millisecond)
+	ticker := time.NewTicker(defaultTaskPollInterval)
 	defer ticker.Stop()
 
 	// Seed revertSkipped from the durable signal so a driver that picks this apply
@@ -646,7 +646,7 @@ func (c *LocalClient) handleAtomicProgressTick(ctx context.Context, eng engine.E
 		ps.consecutiveErrors++
 		logger.Warn("progress check failed",
 			append(apply.MutableLogAttrs(), "error", err, "consecutive_errors", ps.consecutiveErrors)...)
-		if ps.consecutiveErrors >= 10 {
+		if ps.consecutiveErrors >= maxConsecutiveProgressPollErrors {
 			if c.shouldRetryEngineError(err) {
 				logger.Warn("progress polling failed repeatedly, pausing apply for operator retry",
 					"consecutive_errors", ps.consecutiveErrors)