feat(flow): rack-scale decommission workflow with proto mirror sync a…

rest-api/flow/internal/service/server_impl.go

@@ -835,6 +835,12 @@ func (rs *FlowServerImpl) decommissionRackImpl(
 			"target_spec is required",
 		)
 	}
+	if targetSpec.GetComponents() != nil {
+		return nil, status.Error(
+			codes.InvalidArgument,
+			"decommission requires rack targets; component targets are not supported",
+		)
+	}
 
 	info := &operations.DecommissionTaskInfo{
 		RuleID: protobuf.UUIDStringFrom(req.GetRuleId()),

rest-api/flow/internal/task/executor/temporalworkflow/activity/activity.go

@@ -207,14 +207,24 @@ func (a *Activities) DecommissionControl(
 
 // GetDecommissionStatusResult is the result of the GetDecommissionStatus activity.
 type GetDecommissionStatusResult struct {
-	// States maps each component ID to its current raw decommission state
+	// States maps each found component ID to its current raw decommission state
 	// string as returned by the component manager (e.g. "Decommissioning/...",
-	// "Decommissioned", "Failed/...").
+	// "Decommissioned", "Failed/..."). Only IDs that Core returned a record for
+	// are present here.
 	States map[string]string
+	// NotFound holds component IDs that Core has no record of. This happens
+	// when Core removes the resource record as the terminal decommission step,
+	// but can also occur for unknown/mistyped IDs. The workflow logs these at
+	// Warn level and treats them as terminal, since there is no API signal to
+	// distinguish the two cases. A TODO is left to add an explicit
+	// "Decommissioned" state or a found/not-found flag to the Core API.
+	NotFound []string
 }
 
 // GetDecommissionStatus returns the decommission state for target components.
 // This activity is designed to be called repeatedly in a polling loop.
+// Component IDs that Core has no record of are returned in NotFound rather
+// than as empty strings in States, so the caller can distinguish them.
 func (a *Activities) GetDecommissionStatus(
 	ctx context.Context,
 	target common.Target,
@@ -224,12 +234,22 @@ func (a *Activities) GetDecommissionStatus(
 		return nil, err
 	}
 
-	states, err := reader.GetDecommissionStatus(ctx, target)
+	rawStates, err := reader.GetDecommissionStatus(ctx, target)
 	if err != nil {
 		return nil, err
 	}
 
-	return &GetDecommissionStatusResult{States: states}, nil
+	result := &GetDecommissionStatusResult{
+		States: make(map[string]string, len(rawStates)),
+	}
+	for id, state := range rawStates {
+		if state == "" {
+			result.NotFound = append(result.NotFound, id)
+		} else {
+			result.States[id] = state
+		}
+	}
+	return result, nil
 }
 
 // VerifyFirmwareConsistency checks that all target components report the

rest-api/flow/internal/task/executor/temporalworkflow/workflow/actions.go

@@ -647,10 +647,24 @@ func executeDecommissionControlAction(actx actionExecutionContext) error {
 	).Get(ctx, nil)
 }
 
+// maxConsecutiveFailureDuration is the time span over which consecutive
+// GetDecommissionStatus errors must occur before the wait loop aborts.
+// A time-based budget scales with the configured poll interval rather than
+// being coupled to a fixed attempt count: a Core outage that outlasts this
+// window is treated as unrecoverable and the workflow returns an error.
+const maxConsecutiveFailureDuration = 5 * time.Minute
+
 // executeWaitDecommissionedAction polls GetDecommissionStatus until all
 // components reach the "Decommissioned" terminal state. States that begin
-// with "Decommissioning/" are in-progress; any other non-terminal state is
-// treated as an error. Uses config.Timeout and config.PollInterval.
+// with "Decommissioning/" are in-progress; any other state is a hard failure.
+//
+// A component absent from Core's response (state "") is treated as already
+// decommissioned: Core removes the resource record as the final step, so an
+// absent ID is the expected terminal condition rather than an error.
+//
+// Consecutive GetDecommissionStatus failures are counted; after
+// maxConsecutiveStatusFailures the loop aborts rather than spinning until the
+// deadline. Uses config.Timeout and config.PollInterval.
 func executeWaitDecommissionedAction(actx actionExecutionContext) error {
 	ctx := actx.workflowContext
 	target := actx.target
@@ -671,6 +685,7 @@ func executeWaitDecommissionedAction(actx actionExecutionContext) error {
 		Msg("Waiting for decommission to complete")
 
 	deadline := workflow.Now(ctx).Add(timeout)
+	var firstFailureAt time.Time
 
 	for {
 		if workflow.Now(ctx).After(deadline) {
@@ -683,38 +698,81 @@ func executeWaitDecommissionedAction(actx actionExecutionContext) error {
 			return fmt.Errorf("workflow sleep interrupted: %w", err)
 		}
 
+		// Use a short fire-once policy so a hung status call fails quickly
+		// and the poll loop's time-based failure budget controls retries.
+		statusCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
+			ScheduleToCloseTimeout: 30 * time.Second,
+			StartToCloseTimeout:    30 * time.Second,
+			RetryPolicy: &temporal.RetryPolicy{
+				MaximumAttempts: 1,
+			},
+		})
 		var result activity.GetDecommissionStatusResult
 		err := workflow.ExecuteActivity(
-			ctx, activity.NameGetDecommissionStatus, target,
-		).Get(ctx, &result)
+			statusCtx, activity.NameGetDecommissionStatus, target,
+		).Get(statusCtx, &result)
 		if err != nil {
-			log.Warn().Err(err).Msg("Failed to get decommission status, will retry")
+			now := workflow.Now(ctx)
+			if firstFailureAt.IsZero() {
+				firstFailureAt = now
+			}
+			elapsed := now.Sub(firstFailureAt)
+			log.Warn().
+				Err(err).
+				Dur("consecutive_failure_duration", elapsed).
+				Dur("limit", maxConsecutiveFailureDuration).
+				Msg("Failed to get decommission status")
+			if elapsed >= maxConsecutiveFailureDuration {
+				return fmt.Errorf(
+					"aborting: GetDecommissionStatus has been failing for %v: %w",
+					elapsed, err,
+				)
+			}
 			continue
 		}
+		firstFailureAt = time.Time{} // reset on success
+
+		// Treat the result as complete only if Core reported at least one
+		// component (found or absent). An entirely empty response — no states
+		// and no not-found IDs — means the query produced no output and we
+		// should keep polling rather than declaring success.
+		allDecommissioned := len(result.States)+len(result.NotFound) > 0
 
-		allDecommissioned := true
 		for componentID, state := range result.States {
-			if state == "Decommissioned" {
-				continue
-			}
-			if strings.HasPrefix(state, "Decommissioning/") {
+			switch {
+			case state == "Decommissioned":
+				// Terminal success.
+			case strings.HasPrefix(state, "Decommissioning/"):
 				allDecommissioned = false
 				log.Debug().
 					Str("component_id", componentID).
 					Str("state", state).
 					Msg("Component still decommissioning")
-				continue
+			default:
+				return fmt.Errorf(
+					"decommission failed for component %s: reached unexpected state %q",
+					componentID, state,
+				)
 			}
-			// Any other state is unexpected/terminal-error
-			return fmt.Errorf(
-				"decommission failed for component %s: unexpected state %q",
-				componentID, state,
-			)
+		}
+
+		// NotFound components are logged at Warn and treated as terminal.
+		// Core removes the resource record as its final decommission step, so
+		// absence is the expected success signal. However, absence can also mean
+		// an unknown or mistyped ID; without an explicit Core API signal we
+		// cannot tell the two apart.
+		// TODO: add a definitive "Decommissioned" state or found/not-found flag
+		// to the Core API so this ambiguity can be resolved.
+		for _, id := range result.NotFound {
+			log.Warn().
+				Str("component_id", id).
+				Msg("Component absent from Core during decommission poll; treating as terminal (record removed or unknown ID)")
 		}
 
 		if allDecommissioned {
 			log.Info().
-				Int("count", len(result.States)).
+				Int("states_count", len(result.States)).
+				Int("not_found_count", len(result.NotFound)).
 				Msg("All components decommissioned")
 			return nil
 		}