fix(advancer): preserve application failure fences

Author: vfusco

internal/advancer/advancer.go

@@ -81,18 +81,18 @@ func (s *Service) Step(ctx context.Context) (bool, error) {
 	}
 
 	// Update the machine manager with any new or disabled applications
-	err := s.machineManager.UpdateMachines(ctx)
-	if err != nil {
-		return false, err
+	updateErr := s.machineManager.UpdateMachines(ctx)
+	if updateErr != nil && !manager.IsOnlyApplicationFailurePersistenceErrors(updateErr) {
+		return false, updateErr
 	}
 
 	// Get all applications with active machines (returned sorted by ID).
 	apps := s.machineManager.Applications()
 	if len(apps) == 0 {
-		return false, nil
+		return false, updateErr
 	}
 	anyWork := false
-	var errs []error
+	errs := []error{updateErr}
 	for _, app := range apps {
 		hadWork, err := s.stepApp(ctx, app)
 		if err != nil {
@@ -266,13 +266,7 @@ func (s *Service) processInputs(ctx context.Context, app *Application, inputs []
 				"index", input.Index,
 				"error", err)
 
-			if dbErr := appstatus.SetFailed(ctx, s.Logger, s.repository, app, err.Error()); dbErr != nil {
-				s.Logger.Error("Failed to persist FAILED status — machine will be closed "+
-					"but the app status remains unchanged in DB; it may be re-created "+
-					"from the last snapshot on the next tick. If the root cause "+
-					"persists, this may loop.",
-					"application", app.Name, "db_error", dbErr)
-			}
+			s.markApplicationFailed(ctx, app, err.Error())
 
 			// Eagerly close the machine to release the child process.
 			// The app has failed, so no further operations will succeed.
@@ -354,6 +348,20 @@ func (s *Service) processInputs(ctx context.Context, app *Application, inputs []
 	return nil
 }
 
+// markApplicationFailed persists FAILED or installs a local fence when the
+// status write cannot be confirmed. Keeping those operations together prevents
+// a failed application from being handed more work while durability is retried.
+func (s *Service) markApplicationFailed(ctx context.Context, app *Application, reason string) {
+	if err := appstatus.SetFailed(ctx, s.Logger, s.repository, app, reason); err != nil {
+		s.machineManager.FenceApplicationFailure(app, reason)
+		s.Logger.Error(
+			"Could not persist FAILED application status; the application remains fenced until the write is retried",
+			"application", app.Name,
+			"db_error", err,
+		)
+	}
+}
+
 func (s *Service) isEpochLastInput(ctx context.Context, app *Application, input *Input) (bool, error) {
 	if app == nil || input == nil {
 		return false, fmt.Errorf("application and input must not be nil")
@@ -421,10 +429,7 @@ func (s *Service) handleEpochAfterInputsProcessed(ctx context.Context, app *Appl
 			// If the runtime was destroyed (e.g., child process crashed),
 			// mark the app as failed to avoid an infinite retry loop.
 			if errors.Is(err, manager.ErrMachineClosed) {
-				if dbErr := appstatus.SetFailed(ctx, s.Logger, s.repository, app, err.Error()); dbErr != nil {
-					s.Logger.Error("Failed to persist FAILED status for crashed machine",
-						"application", app.Name, "db_error", dbErr)
-				}
+				s.markApplicationFailed(ctx, app, err.Error())
 			}
 			return fmt.Errorf("failed to get outputs proof from machine: %w", err)
 		}

internal/advancer/advancer_test.go

@@ -19,6 +19,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/cartesi/rollups-node/internal/appstatus"
 	"github.com/cartesi/rollups-node/internal/manager"
 	. "github.com/cartesi/rollups-node/internal/model"
 	"github.com/cartesi/rollups-node/internal/repository"
@@ -118,6 +119,10 @@ func (s *AdvancerSuite) TestServiceInterface() {
 		// Test service interface methods
 		require.True(advancer.Alive())
 		require.True(advancer.Ready())
+		machineManager.PendingApplicationFailures = true
+		require.False(advancer.Ready())
+		machineManager.PendingApplicationFailures = false
+		require.True(advancer.Ready())
 		require.Empty(advancer.Reload())
 		require.Equal(advancer.Name, advancer.String())
 
@@ -224,6 +229,40 @@ func (s *AdvancerSuite) TestStep() {
 		require.Contains(err.Error(), "update machines error")
 	})
 
+	s.Run("Error/UnconfirmedApplicationFailureStatus", func() {
+		require := s.Require()
+		persistenceErr := &manager.ApplicationFailurePersistenceError{
+			ApplicationID: 7,
+			WriteErr:      errors.New("status write unavailable"),
+			ReadErr:       errors.New("status read unavailable"),
+		}
+		healthy := newMockMachine(8)
+		machineManager := &MockMachineManager{
+			Map: map[int64]*MockMachineInstance{
+				healthy.Application.ID: newMockInstance(healthy),
+			},
+			UpdateMachinesError:        persistenceErr,
+			PendingApplicationFailures: true,
+		}
+		repo := &MockRepository{
+			GetEpochsReturn: map[common.Address][]*Epoch{
+				healthy.Application.IApplicationAddress: {{Index: 0, Status: EpochStatus_Open}},
+			},
+			GetInputsReturn: map[common.Address][]*Input{
+				healthy.Application.IApplicationAddress: {
+					newInput(healthy.Application.ID, 0, 0, marshal(randomAdvanceResult(0))),
+				},
+			},
+		}
+		advancer, err := newMockAdvancerService(machineManager, repo)
+		require.NoError(err)
+
+		_, err = advancer.Step(context.Background())
+		require.ErrorIs(err, manager.ErrApplicationFailureNotDurable)
+		require.Len(repo.StoredResults, 1, "the healthy application must still advance")
+		require.False(advancer.Ready())
+	})
+
 	s.Run("Error/GetInputs", func() {
 		require := s.Require()
 		env := s.setupOneApp()
@@ -295,6 +334,48 @@ func (s *AdvancerSuite) TestStep() {
 		// app2's input was processed despite app1's failure
 		require.Len(repo.StoredResults, 1)
 	})
+
+	s.Run("LiveCycleLimitWriteFailureFencesAppWithoutBlockingHealthySibling", func() {
+		require := s.Require()
+		mm := newMockMachineManager()
+		limited := newMockMachine(1)
+		healthy := newMockMachine(2)
+		limitErr := fmt.Errorf(
+			"advance execution reached configured cycle limit: %w",
+			pkgmachine.ErrReachedLimitMcycle,
+		)
+		limited.AdvanceError = limitErr
+		mm.Map[limited.Application.ID] = newMockInstance(limited)
+		mm.Map[healthy.Application.ID] = newMockInstance(healthy)
+		repo := &MockRepository{
+			GetEpochsReturn: map[common.Address][]*Epoch{
+				limited.Application.IApplicationAddress: {{Index: 0, Status: EpochStatus_Open}},
+				healthy.Application.IApplicationAddress: {{Index: 0, Status: EpochStatus_Open}},
+			},
+			GetInputsReturn: map[common.Address][]*Input{
+				limited.Application.IApplicationAddress: {
+					newInput(limited.Application.ID, 0, 0, []byte("limited input")),
+				},
+				healthy.Application.IApplicationAddress: {
+					newInput(healthy.Application.ID, 0, 0, marshal(randomAdvanceResult(0))),
+				},
+			},
+			UpdateApplicationStatusError: errors.New("FAILED write unavailable"),
+		}
+		svc, err := newMockAdvancerService(mm, repo)
+		require.NoError(err)
+
+		_, err = svc.Step(context.Background())
+
+		require.ErrorIs(err, pkgmachine.ErrReachedLimitMcycle)
+		require.Equal(
+			appstatus.NormalizeReason(limitErr.Error()),
+			mm.RecordedApplicationFailures[limited.Application.ID],
+		)
+		require.False(svc.Ready(), "an unconfirmed FAILED write must fail readiness immediately")
+		require.Len(repo.StoredResults, 1, "the healthy sibling must still advance")
+		require.Equal(healthy.Application.ID, repo.StoredAppIDs[0])
+	})
 }
 
 func (s *AdvancerSuite) TestGetUnprocessedInputs() {
@@ -601,6 +682,11 @@ func (s *AdvancerSuite) TestContextCancellation() {
 		// The expired context prevents the immediate database write, so the
 		// application failure is queued for a later status-write retry.
 		require.Zero(env.repo.ApplicationStatusUpdates)
+		require.Equal(
+			context.DeadlineExceeded.Error(),
+			env.mm.RecordedApplicationFailures[env.app.Application.ID],
+		)
+		require.True(env.mm.PendingApplicationFailures)
 		require.Equal(1, env.mm.Map[env.app.Application.ID].closeCalls)
 		require.True(logs.contains(slog.LevelError, "Error executing advance"))
 		require.False(logs.contains(
@@ -893,6 +979,23 @@ func (s *AdvancerSuite) TestHandleEpochAfterInputsProcessed() {
 		require.Equal(ApplicationStatus_Failed, env.repo.LastApplicationStatus)
 	})
 
+	s.Run("EmptyEpochIndex0ErrMachineClosedWriteFailureQueuesDurableFence", func() {
+		require := s.Require()
+		env := s.setupOneApp()
+		env.app.OutputsProofError = manager.ErrMachineClosed
+		env.repo.UpdateApplicationStatusError = errors.New("FAILED write unavailable")
+		epoch := &Epoch{Index: 0, Status: EpochStatus_Closed, InputIndexLowerBound: 0, InputIndexUpperBound: 0}
+
+		err := env.service.handleEpochAfterInputsProcessed(context.Background(), env.app.Application, epoch)
+
+		require.ErrorIs(err, manager.ErrMachineClosed)
+		require.Equal(
+			appstatus.NormalizeReason(manager.ErrMachineClosed.Error()),
+			env.mm.RecordedApplicationFailures[env.app.Application.ID],
+		)
+		require.False(env.service.Ready())
+	})
+
 	s.Run("EmptyEpochIndexGt0RepeatsPreviousProof", func() {
 		require := s.Require()
 		env := s.setupOneApp()
@@ -1957,13 +2060,16 @@ func newMockInstance(impl *MockMachineImpl) *MockMachineInstance {
 // ------------------------------------------------------------------------------------------------
 
 type MockMachineManager struct {
-	Map                 map[int64]*MockMachineInstance
-	UpdateMachinesError error
+	Map                         map[int64]*MockMachineInstance
+	UpdateMachinesError         error
+	PendingApplicationFailures  bool
+	RecordedApplicationFailures map[int64]string
 }
 
 func newMockMachineManager() *MockMachineManager {
 	return &MockMachineManager{
-		Map: map[int64]*MockMachineInstance{},
+		Map:                         map[int64]*MockMachineInstance{},
+		RecordedApplicationFailures: map[int64]string{},
 	}
 }
 
@@ -1979,6 +2085,14 @@ func (mock *MockMachineManager) UpdateMachines(ctx context.Context) error {
 	return mock.UpdateMachinesError
 }
 
+func (mock *MockMachineManager) FenceApplicationFailure(app *Application, reason string) {
+	if mock.RecordedApplicationFailures == nil {
+		mock.RecordedApplicationFailures = map[int64]string{}
+	}
+	mock.RecordedApplicationFailures[app.ID] = appstatus.NormalizeReason(reason)
+	mock.PendingApplicationFailures = true
+}
+
 func (mock *MockMachineManager) Applications() []*Application {
 	apps := make([]*Application, 0, len(mock.Map))
 	for _, v := range mock.Map {
@@ -1993,6 +2107,10 @@ func (mock *MockMachineManager) HasMachine(appID int64) bool {
 	return exists
 }
 
+func (mock *MockMachineManager) HasPendingApplicationFailures() bool {
+	return mock.PendingApplicationFailures
+}
+
 func (mock *MockMachineManager) Close() error {
 	return nil
 }

internal/advancer/determinism_test.go

@@ -229,6 +229,12 @@ func testDeterminismCallerDeadlines(
 		require.Equal(t, uint64(1), harness.instance.ProcessedInputs())
 		require.Zero(t, repo.ApplicationStatusUpdates,
 			"the expired context prevents the immediate FAILED status write")
+		require.True(t, harness.provider.HasPendingApplicationFailures())
+		require.Equal(
+			t,
+			context.DeadlineExceeded.Error(),
+			harness.provider.failureReason(harness.app.ID),
+		)
 		require.Equal(t, predecessor, harness.factory.base.snapshot())
 		require.True(t, harness.factory.base.isClosed(),
 			"a timed-out advance must close the changed live machine")
@@ -894,7 +900,7 @@ func (p *determinismMachineProvider) Applications() []*model.Application {
 
 func (p *determinismMachineProvider) UpdateMachines(context.Context) error { return nil }
 
-func (p *determinismMachineProvider) RecordApplicationFailure(app *model.Application, reason string) {
+func (p *determinismMachineProvider) FenceApplicationFailure(app *model.Application, reason string) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 	if p.failures == nil {

internal/advancer/service.go

@@ -113,8 +113,13 @@ func Create(ctx context.Context, c *CreateInfo) (*Service, error) {
 }
 
 // Service interface implementation
-func (s *Service) Alive() bool     { return true }
-func (s *Service) Ready() bool     { return true }
+func (s *Service) Alive() bool { return true }
+func (s *Service) Ready() bool {
+	// This is a local fail-closed signal while application-failure durability
+	// is unresolved. It cannot atomically revoke work already selected by a
+	// separate process before that durable status becomes visible.
+	return s.machineManager != nil && !s.machineManager.HasPendingApplicationFailures()
+}
 func (s *Service) Reload() []error { return nil }
 func (s *Service) Tick() []error {
 	hadWork, err := s.Step(s.Context)

internal/manager/types.go

@@ -44,9 +44,17 @@ type MachineProvider interface {
 	// UpdateMachines refreshes the list of machines
 	UpdateMachines(ctx context.Context) error
 
+	// FenceApplicationFailure fences an application whose initial FAILED
+	// status write could not be confirmed. It queues a later durability retry
+	// without duplicating the initial repository write.
+	FenceApplicationFailure(app *Application, reason string)
+
 	// HasMachine checks if a machine exists for the given application ID
 	HasMachine(appID int64) bool
 
+	// HasPendingApplicationFailures reports an unresolved durable-status fence.
+	HasPendingApplicationFailures() bool
+
 	// Close shuts down all machine instances and releases resources
 	Close() error
 }