fix: remove backoff retry from coordinator metadata status operations

The status resource was using backoff.NewExponentialBackOff() from the
cenkalti library directly, which has a default MaxElapsedTime of 15
minutes and is not context-aware. This caused several issues:

- Held mutex lock during entire retry loop (up to 15 minutes)
- Not context-aware, so retries couldn't be cancelled on shutdown
- Errors silently discarded when retries exhausted
- Inconsistent with the rest of the codebase which uses oxiatime.NewBackOff(ctx)

The callers (shard controllers, elections, periodic tasks) already have
their own retry mechanisms, making the status resource retry redundant.

This change removes the backoff retry, returns errors to callers, and
lets them handle failures appropriately.

Author: mattisonchao

oxiad/coordinator/balancer/scheduler_test.go

@@ -43,17 +43,22 @@ func (m *mockStatusResource) LoadWithVersion() (*model.ClusterStatus, metadata.V
 	return m.status, "0"
 }
 
-func (m *mockStatusResource) Swap(_ *model.ClusterStatus, _ metadata.Version) bool {
-	return true
+func (m *mockStatusResource) Swap(_ *model.ClusterStatus, _ metadata.Version) (bool, error) {
+	return true, nil
 }
 
-func (m *mockStatusResource) Update(newStatus *model.ClusterStatus) {
+func (m *mockStatusResource) Update(newStatus *model.ClusterStatus) error {
 	m.status = newStatus
+	return nil
 }
 
-func (m *mockStatusResource) UpdateShardMetadata(_ string, _ int64, _ model.ShardMetadata) {}
+func (m *mockStatusResource) UpdateShardMetadata(_ string, _ int64, _ model.ShardMetadata) error {
+	return nil
+}
 
-func (m *mockStatusResource) DeleteShardMetadata(_ string, _ int64) {}
+func (m *mockStatusResource) DeleteShardMetadata(_ string, _ int64) error {
+	return nil
+}
 
 func (m *mockStatusResource) IsReady(_ *model.ClusterConfig) bool {
 	return true

oxiad/coordinator/controller/shard_controller.go

@@ -427,7 +427,9 @@ func (s *shardController) deleteShard() error {
 		)
 	}
 
-	s.statusResource.DeleteShardMetadata(s.namespace, s.shard)
+	if err := s.statusResource.DeleteShardMetadata(s.namespace, s.shard); err != nil {
+		return err
+	}
 	s.eventListener.ShardDeleted(s.shard)
 	return s.close()
 }
@@ -508,7 +510,10 @@ func (s *shardController) handlePeriodicTasks() {
 	}
 
 	// Update the shard status
-	s.statusResource.UpdateShardMetadata(s.namespace, s.shard, mutShardMeta)
+	if err := s.statusResource.UpdateShardMetadata(s.namespace, s.shard, mutShardMeta); err != nil {
+		s.log.Warn("Failed to update shard metadata", "error", err)
+		return
+	}
 	s.metadata.Store(mutShardMeta)
 }
 

oxiad/coordinator/controller/shard_controller_election.go

@@ -434,7 +434,9 @@ func (e *ShardElection) start() (model.Server, error) {
 		metadata.Term++
 		metadata.Ensemble = e.refreshedEnsemble(metadata.Ensemble)
 	})
-	e.statusResource.UpdateShardMetadata(e.namespace, e.shard, mutShardMeta)
+	if err := e.statusResource.UpdateShardMetadata(e.namespace, e.shard, mutShardMeta); err != nil {
+		return model.Server{}, err
+	}
 
 	if e.changeEnsembleAction != nil {
 		e.prepareIfChangeEnsemble(&mutShardMeta)
@@ -484,7 +486,9 @@ func (e *ShardElection) start() (model.Server, error) {
 	leader := mutShardMeta.Leader
 	leaderEntry := candidatesStatus[*leader]
 
-	e.statusResource.UpdateShardMetadata(e.namespace, e.shard, mutShardMeta)
+	if err = e.statusResource.UpdateShardMetadata(e.namespace, e.shard, mutShardMeta); err != nil {
+		return model.Server{}, err
+	}
 	e.meta.Store(mutShardMeta)
 	if e.eventListener != nil {
 		e.eventListener.LeaderElected(e.shard, newLeader, maps.Keys(followers))

oxiad/coordinator/controller/split_controller.go

@@ -215,7 +215,7 @@ func (sc *SplitController) currentPhase() *model.SplitPhase {
 }
 
 // updatePhase atomically updates the split phase on both parent and children.
-func (sc *SplitController) updatePhase(newPhase model.SplitPhase) {
+func (sc *SplitController) updatePhase(newPhase model.SplitPhase) error {
 	status := sc.statusResource.Load()
 	cloned := status.Clone()
 
@@ -235,7 +235,7 @@ func (sc *SplitController) updatePhase(newPhase model.SplitPhase) {
 		}
 	}
 
-	sc.statusResource.Update(cloned)
+	return sc.statusResource.Update(cloned)
 }
 
 // runBootstrap validates preconditions, fences child ensemble members, elects
@@ -285,13 +285,14 @@ func (sc *SplitController) runBootstrap() error {
 			childLeaders[childId] = childMeta.Leader.Internal
 		}
 	}
-	sc.updateParentMeta(func(meta *model.ShardMetadata) {
+	if err := sc.updateParentMeta(func(meta *model.ShardMetadata) {
 		meta.Split.ParentTermAtBootstrap = parentTerm
 		meta.Split.ChildLeadersAtBootstrap = childLeaders
-	})
+	}); err != nil {
+		return err
+	}
 
-	sc.updatePhase(model.SplitPhaseCatchUp)
-	return nil
+	return sc.updatePhase(model.SplitPhaseCatchUp)
 }
 
 // fenceAndElectChild fences a child shard's ensemble and elects a leader.
@@ -318,11 +319,13 @@ func (sc *SplitController) fenceAndElectChild(childId int64) error {
 
 	childLeader := sc.pickLeader(headEntries)
 
-	sc.updateChildMeta(childId, func(meta *model.ShardMetadata) {
+	if err := sc.updateChildMeta(childId, func(meta *model.ShardMetadata) {
 		meta.Term = childTerm
 		meta.Leader = &childLeader
 		meta.Status = model.ShardStatusSteadyState
-	})
+	}); err != nil {
+		return err
+	}
 
 	// Elect the child leader so it replicates to its followers immediately.
 	// Without this, only the single child leader node has the data.
@@ -421,8 +424,7 @@ func (sc *SplitController) runCatchUp() error {
 		}
 		if caughtUp {
 			sc.log.Info("All children caught up")
-			sc.updatePhase(model.SplitPhaseCutover)
-			return nil
+			return sc.updatePhase(model.SplitPhaseCutover)
 		}
 	}
 }
@@ -443,14 +445,18 @@ func (sc *SplitController) checkObserverCursorsStale() (bool, error) {
 			slog.Int64("bootstrap-term", parentMeta.Split.ParentTermAtBootstrap),
 			slog.Int64("current-term", parentMeta.Term),
 		)
-		sc.updatePhase(model.SplitPhaseBootstrap)
+		if err := sc.updatePhase(model.SplitPhaseBootstrap); err != nil {
+			return false, err
+		}
 		return true, nil
 	}
 
 	// Child leader election: the observer cursor targets the old (dead) leader.
 	// Remove the stale cursor and fall back to Bootstrap to re-add.
 	if sc.removeStaleChildObservers(parentMeta) {
-		sc.updatePhase(model.SplitPhaseBootstrap)
+		if err := sc.updatePhase(model.SplitPhaseBootstrap); err != nil {
+			return false, err
+		}
 		return true, nil
 	}
 
@@ -562,11 +568,13 @@ func (sc *SplitController) runCutover() error {
 	)
 
 	// Update parent term in metadata
-	sc.updateParentMeta(func(meta *model.ShardMetadata) {
+	if err := sc.updateParentMeta(func(meta *model.ShardMetadata) {
 		meta.Term = newParentTerm
 		meta.Leader = nil
 		meta.Status = model.ShardStatusElection
-	})
+	}); err != nil {
+		return err
+	}
 
 	// Step 2: Wait for children to commit parentFinalOffset.
 	// Children were already elected leader in Bootstrap, so commitOffset
@@ -587,14 +595,18 @@ func (sc *SplitController) runCutover() error {
 	// Step 4: Clear split metadata from children and mark parent for deletion.
 	// Children are now independent shards.
 	for _, childId := range []int64{sc.leftChildId, sc.rightChildId} {
-		sc.updateChildMeta(childId, func(meta *model.ShardMetadata) {
+		if err := sc.updateChildMeta(childId, func(meta *model.ShardMetadata) {
 			meta.Split = nil
-		})
+		}); err != nil {
+			return err
+		}
 	}
 
-	sc.updateParentMeta(func(meta *model.ShardMetadata) {
+	if err := sc.updateParentMeta(func(meta *model.ShardMetadata) {
 		meta.Status = model.ShardStatusDeleting
-	})
+	}); err != nil {
+		return err
+	}
 
 	// Step 5: Notify the coordinator. This triggers the parent shard
 	// controller's DeleteShard (which retries indefinitely with backoff)
@@ -603,9 +615,11 @@ func (sc *SplitController) runCutover() error {
 
 	// Clear split metadata from parent — the split controller's job is done.
 	// The parent shard controller handles the actual deletion.
-	sc.updateParentMeta(func(meta *model.ShardMetadata) {
+	if err := sc.updateParentMeta(func(meta *model.ShardMetadata) {
 		meta.Split = nil
-	})
+	}); err != nil {
+		return err
+	}
 
 	return nil
 }
@@ -650,13 +664,19 @@ func (sc *SplitController) abort() {
 
 	// Delete child shards from status.
 	for _, childId := range []int64{sc.leftChildId, sc.rightChildId} {
-		sc.statusResource.DeleteShardMetadata(sc.namespace, childId)
+		if err := sc.statusResource.DeleteShardMetadata(sc.namespace, childId); err != nil {
+			sc.log.Warn("Failed to delete child shard metadata during abort",
+				slog.Int64("child-shard", childId),
+				slog.Any("error", err))
+		}
 	}
 
 	// Clear parent split metadata.
-	sc.updateParentMeta(func(meta *model.ShardMetadata) {
+	if err := sc.updateParentMeta(func(meta *model.ShardMetadata) {
 		meta.Split = nil
-	})
+	}); err != nil {
+		sc.log.Warn("Failed to clear parent split metadata during abort", slog.Any("error", err))
+	}
 
 	sc.log.Info("Split aborted, parent restored")
 
@@ -684,23 +704,24 @@ func (sc *SplitController) loadShardMeta(shardId int64) *model.ShardMetadata {
 	return &cloned
 }
 
-func (sc *SplitController) updateParentMeta(fn func(meta *model.ShardMetadata)) {
-	sc.updateShardMeta(sc.parentShardId, fn)
+func (sc *SplitController) updateParentMeta(fn func(meta *model.ShardMetadata)) error {
+	return sc.updateShardMeta(sc.parentShardId, fn)
 }
 
-func (sc *SplitController) updateChildMeta(childId int64, fn func(meta *model.ShardMetadata)) {
-	sc.updateShardMeta(childId, fn)
+func (sc *SplitController) updateChildMeta(childId int64, fn func(meta *model.ShardMetadata)) error {
+	return sc.updateShardMeta(childId, fn)
 }
 
-func (sc *SplitController) updateShardMeta(shardId int64, fn func(meta *model.ShardMetadata)) {
+func (sc *SplitController) updateShardMeta(shardId int64, fn func(meta *model.ShardMetadata)) error {
 	status := sc.statusResource.Load()
 	cloned := status.Clone()
 	ns := cloned.Namespaces[sc.namespace]
 	if meta, exists := ns.Shards[shardId]; exists {
 		fn(&meta)
 		ns.Shards[shardId] = meta
-		sc.statusResource.Update(cloned)
+		return sc.statusResource.Update(cloned)
 	}
+	return nil
 }
 
 // fenceEnsemble sends NewTerm to all ensemble members and returns the
@@ -860,11 +881,13 @@ func (sc *SplitController) reelectChild(childId int64) error {
 	}
 
 	// Update child metadata
-	sc.updateChildMeta(childId, func(meta *model.ShardMetadata) {
+	if err := sc.updateChildMeta(childId, func(meta *model.ShardMetadata) {
 		meta.Term = newTerm
 		meta.Leader = &newLeader
 		meta.Status = model.ShardStatusSteadyState
-	})
+	}); err != nil {
+		return err
+	}
 
 	sc.log.Info("Child re-elected in clean term",
 		slog.Int64("child-shard", childId),

oxiad/coordinator/coordinator.go

@@ -173,7 +173,13 @@ func (c *coordinator) ConfigChanged(newConfig *model.ClusterConfig) {
 	var shardsToDelete []int64
 	for {
 		clusterStatus, shardsToAdd, shardsToDelete = util.ApplyClusterChanges(newConfig, currentStatus, c.selectNewEnsemble)
-		if !c.statusResource.Swap(clusterStatus, version) {
+		swapped, err := c.statusResource.Swap(clusterStatus, version)
+		if err != nil {
+			c.Warn("Failed to swap cluster status", slog.Any("error", err))
+			currentStatus, version = c.statusResource.LoadWithVersion()
+			continue
+		}
+		if !swapped {
 			currentStatus, version = c.statusResource.LoadWithVersion()
 			continue
 		}
@@ -563,7 +569,9 @@ func (c *coordinator) InitiateSplit(namespace string, parentShardId int64, split
 	}
 
 	// Persist
-	c.statusResource.Update(cloned)
+	if err = c.statusResource.Update(cloned); err != nil {
+		return 0, 0, errors.Wrap(err, "failed to persist split status")
+	}
 
 	c.Info("Split initiated",
 		slog.Int64("parent-shard", parentShardId),
@@ -777,7 +785,9 @@ func NewCoordinator(meta metadata.Provider,
 
 		clusterStatus, _, _ = util.ApplyClusterChanges(clusterConfig, model.NewClusterStatus(), c.selectNewEnsemble)
 
-		c.statusResource.Update(clusterStatus)
+		if err = c.statusResource.Update(clusterStatus); err != nil {
+			return nil, nil, errors.Wrap(err, "failed to persist initial cluster status")
+		}
 	} else {
 		c.Info("Checking cluster config", slog.Any("clusterConfig", clusterConfig))
 
@@ -787,7 +797,9 @@ func NewCoordinator(meta metadata.Provider,
 			clusterStatus, c.selectNewEnsemble)
 
 		if len(shardsToAdd) > 0 || len(shardsToDelete) > 0 {
-			c.statusResource.Update(clusterStatus)
+			if err = c.statusResource.Update(clusterStatus); err != nil {
+				return nil, nil, errors.Wrap(err, "failed to persist cluster status changes")
+			}
 		}
 	}