feat(flow): add event-rule execution scheduler

rest-api/common/pkg/util/converter.go

@@ -46,6 +46,29 @@ func GetPtr[T any](v T) *T {
 	return &v
 }
 
+// GetPtrIfNotZero returns nil for a type's zero value or a pointer to a copy
+// of v otherwise.
+func GetPtrIfNotZero[T comparable](v T) *T {
+	var zero T
+	if v == zero {
+		return nil
+	}
+
+	return &v
+}
+
+// GetValueOrZero returns a type's zero value for nil or the dereferenced value
+// otherwise.
+func GetValueOrZero[T any](v *T) T {
+	if v == nil {
+		var zero T
+
+		return zero
+	}
+
+	return *v
+}
+
 // StrPtrToProtoTimePtr converts a string pointer to a protobuf timestamp pointer
 func StrPtrToProtoTimePtr(s *string) *timestamppb.Timestamp {
 	if s == nil {

rest-api/common/pkg/util/converter_test.go

@@ -97,3 +97,47 @@ func TestGetPtr(t *testing.T) {
 		}
 	})
 }
+
+func TestGetPtrIfNotZero(t *testing.T) {
+	tests := []struct {
+		name    string
+		value   string
+		wantNil bool
+	}{
+		{name: "zero value", wantNil: true},
+		{name: "non-zero value", value: "test"},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			got := GetPtrIfNotZero(test.value)
+
+			if test.wantNil {
+				require.Nil(t, got)
+
+				return
+			}
+
+			require.NotNil(t, got)
+			require.Equal(t, test.value, *got)
+		})
+	}
+}
+
+func TestGetValueOrZero(t *testing.T) {
+	tests := []struct {
+		name  string
+		value *string
+		want  string
+	}{
+		{name: "nil"},
+		{name: "pointer to zero value", value: GetPtr("")},
+		{name: "pointer to non-zero value", value: GetPtr("test"), want: "test"},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			require.Equal(t, test.want, GetValueOrZero(test.value))
+		})
+	}
+}

rest-api/flow/internal/converter/dao/converter.go

@@ -12,6 +12,7 @@ import (
 	"github.com/google/uuid"
 
 	"github.com/NVIDIA/infra-controller/rest-api/common/pkg/credential"
+	cutil "github.com/NVIDIA/infra-controller/rest-api/common/pkg/util"
 	"github.com/NVIDIA/infra-controller/rest-api/flow/internal/db/model"
 	"github.com/NVIDIA/infra-controller/rest-api/flow/internal/nicoapi"
 	"github.com/NVIDIA/infra-controller/rest-api/flow/internal/operation"
@@ -92,10 +93,6 @@ func ComponentFrom(dao model.Component) *component.Component {
 		bmcsByType[t] = append(bmcsByType[t], BMCFrom(bd))
 	}
 
-	var componentID string
-	if dao.ComponentID != nil {
-		componentID = *dao.ComponentID
-	}
 	var nvlDomainID uuid.UUID
 	if dao.Rack != nil && dao.Rack.NVLDomainID != uuid.Nil {
 		nvlDomainID = dao.Rack.NVLDomainID
@@ -118,7 +115,7 @@ func ComponentFrom(dao model.Component) *component.Component {
 			HostID:    dao.HostID,
 		},
 		BmcsByType:  bmcsByType,
-		ComponentID: componentID,
+		ComponentID: cutil.GetValueOrZero(dao.ComponentID),
 		RackID:      dao.RackID,
 		NVLDomainID: nvlDomainID,
 		PowerState:  powerStateFromDAO(dao.PowerState),
@@ -294,10 +291,7 @@ func ComponentTo(c *component.Component, rackID uuid.UUID) *model.Component {
 		TrayIndex:       c.Position.TrayIndex,
 		HostID:          c.Position.HostID,
 		RackID:          rackID,
-	}
-
-	if c.ComponentID != "" {
-		compDAO.ComponentID = &c.ComponentID
+		ComponentID:     cutil.GetPtrIfNotZero(c.ComponentID),
 	}
 
 	for _, t := range devicetypes.BMCTypes() {

rest-api/flow/internal/converter/dao/event.go

@@ -35,7 +35,6 @@ func EventTo(event *eventrule.Event) (*dbmodel.Event, error) {
 		Observations:    event.Observations,
 		CreatedAt:       event.CreatedAt,
 		LastObservedAt:  event.LastObservedAt,
-		PlannedAt:       event.PlannedAt,
 	}, nil
 }
 
@@ -64,7 +63,6 @@ func EventFrom(persisted *dbmodel.Event) (*eventrule.Event, error) {
 		Observations:    persisted.Observations,
 		CreatedAt:       persisted.CreatedAt,
 		LastObservedAt:  persisted.LastObservedAt,
-		PlannedAt:       persisted.PlannedAt,
 	}
 
 	if err := event.Validate(); err != nil {

rest-api/flow/internal/converter/dao/event_action_execution.go

@@ -5,8 +5,8 @@ package dao
 
 import (
 	"fmt"
-	"time"
 
+	cutil "github.com/NVIDIA/infra-controller/rest-api/common/pkg/util"
 	dbmodel "github.com/NVIDIA/infra-controller/rest-api/flow/internal/db/model"
 	"github.com/NVIDIA/infra-controller/rest-api/flow/internal/eventrule"
 	eventrulecodec "github.com/NVIDIA/infra-controller/rest-api/flow/internal/eventrule/codec"
@@ -19,39 +19,27 @@ func EventActionExecutionTo(
 	if err := execution.Validate(); err != nil {
 		return nil, err
 	}
+
 	plan, err := eventrulecodec.MarshalExecutionPlan(execution.Plan)
 	if err != nil {
 		return nil, fmt.Errorf("encode execution plan: %w", err)
 	}
 
-	var nextAttemptAt *time.Time
-	if !execution.NextAttemptAt.IsZero() {
-		next := execution.NextAttemptAt
-		nextAttemptAt = &next
-	}
-	var reason *string
-	if execution.Reason != eventrule.ExecutionReasonNone {
-		value := string(execution.Reason)
-		reason = &value
-	}
-	var statusMessage *string
-	if execution.StatusMessage != "" {
-		value := execution.StatusMessage
-		statusMessage = &value
-	}
 	return &dbmodel.EventActionExecution{
 		ID:            execution.ID,
 		EventID:       execution.EventID,
 		ActionName:    execution.ActionName,
 		ActionType:    string(execution.Plan.Type()),
 		Plan:          plan,
 		Status:        string(execution.Status),
-		Reason:        reason,
+		Reason:        cutil.GetPtrIfNotZero(string(execution.Reason)),
 		Attempts:      execution.Attempts,
-		StatusMessage: statusMessage,
+		ClaimToken:    cutil.GetPtrIfNotZero(execution.ClaimToken),
+		ClaimOwner:    cutil.GetPtrIfNotZero(execution.ClaimOwner),
+		StatusMessage: cutil.GetPtrIfNotZero(execution.StatusMessage),
 		CreatedAt:     execution.CreatedAt,
 		UpdatedAt:     execution.UpdatedAt,
-		NextAttemptAt: nextAttemptAt,
+		NextAttemptAt: cutil.GetPtrIfNotZero(execution.NextAttemptAt),
 	}, nil
 }
 
@@ -63,18 +51,6 @@ func EventActionExecutionFrom(
 		return nil, nil
 	}
 
-	var nextAttemptAt time.Time
-	if persisted.NextAttemptAt != nil {
-		nextAttemptAt = *persisted.NextAttemptAt
-	}
-	var reason eventrule.ExecutionReason
-	if persisted.Reason != nil {
-		reason = eventrule.ExecutionReason(*persisted.Reason)
-	}
-	var statusMessage string
-	if persisted.StatusMessage != nil {
-		statusMessage = *persisted.StatusMessage
-	}
 	plan, err := eventrulecodec.UnmarshalExecutionPlan(persisted.Plan)
 	if err != nil {
 		return nil, fmt.Errorf(
@@ -83,6 +59,7 @@ func EventActionExecutionFrom(
 			err,
 		)
 	}
+
 	if string(plan.Type()) != persisted.ActionType {
 		return nil, fmt.Errorf(
 			"%w: action type %q does not match plan type %q",
@@ -91,25 +68,30 @@ func EventActionExecutionFrom(
 			plan.Type(),
 		)
 	}
+
 	execution := &eventrule.Execution{
 		ExecutionState: eventrule.ExecutionState{
 			ExecutionStatusDetails: eventrule.ExecutionStatusDetails{
 				Status:        eventrule.ExecutionStatus(persisted.Status),
-				Reason:        reason,
-				StatusMessage: statusMessage,
+				Reason:        eventrule.ExecutionReason(cutil.GetValueOrZero(persisted.Reason)),
+				StatusMessage: cutil.GetValueOrZero(persisted.StatusMessage),
 			},
-			NextAttemptAt: nextAttemptAt,
+			NextAttemptAt: cutil.GetValueOrZero(persisted.NextAttemptAt),
 		},
 		ID:         persisted.ID,
 		EventID:    persisted.EventID,
 		ActionName: persisted.ActionName,
 		Plan:       plan,
 		Attempts:   persisted.Attempts,
+		ClaimToken: cutil.GetValueOrZero(persisted.ClaimToken),
+		ClaimOwner: cutil.GetValueOrZero(persisted.ClaimOwner),
 		CreatedAt:  persisted.CreatedAt,
 		UpdatedAt:  persisted.UpdatedAt,
 	}
+
 	if err := execution.Validate(); err != nil {
 		return nil, fmt.Errorf("%w: %w", eventrule.ErrInvalidPersistedExecution, err)
 	}
+
 	return execution, nil
 }

rest-api/flow/internal/converter/dao/event_action_execution_test.go

@@ -19,29 +19,90 @@ func TestEventActionExecutionRoundTrip(t *testing.T) {
 	base, err := eventrule.NewExecution(uuid.New(), "notify", &eventrule.NoopPlan{Reason: "test"}, now)
 	require.NoError(t, err)
 
-	tests := map[string]eventrule.ExecutionResult{
-		"completed": eventrule.CompletedExecutionResult(),
-		"skipped":   eventrule.SkippedExecutionResult(eventrule.ExecutionReasonNoTargets),
-		"deferred":  eventrule.DeferredExecutionResult(eventrule.ExecutionReasonAttemptFailed, "temporary", time.Minute),
-		"failed":    eventrule.FailedExecutionResult("terminal"),
+	tests := map[string]struct {
+		claim       bool
+		result      *eventrule.ExecutionResult
+		activeClaim bool
+	}{
+		"pending": {},
+		"running": {
+			claim:       true,
+			activeClaim: true,
+		},
+		"completed": {
+			claim:  true,
+			result: resultPointer(eventrule.CompletedExecutionResult()),
+		},
+		"deferred": {
+			claim: true,
+			result: resultPointer(eventrule.DeferredExecutionResult(
+				eventrule.ExecutionReasonAttemptFailed,
+				"temporary",
+				time.Minute,
+			)),
+		},
+		"interrupted": {
+			claim: true,
+			result: resultPointer(eventrule.DeferredExecutionResult(
+				eventrule.ExecutionReasonAttemptInterrupted,
+				"interrupted",
+				time.Minute,
+			)),
+		},
+		"failed": {
+			claim:  true,
+			result: resultPointer(eventrule.FailedExecutionResult("terminal")),
+		},
 	}
-	for name, result := range tests {
+
+	for name, test := range tests {
 		t.Run(name, func(t *testing.T) {
 			execution := base.Clone()
-			require.NoError(t, execution.TransitionTo(result, now.Add(time.Second)))
+			token := uuid.New()
+
+			if test.claim {
+				require.NoError(t, execution.Claim("scheduler-1", token, now.Add(time.Second)))
+			}
+			if test.result != nil {
+				require.NoError(
+					t,
+					execution.TransitionClaimedTo(token, *test.result, now.Add(2*time.Second)),
+				)
+			}
+
 			persisted, err := EventActionExecutionTo(&execution)
 			require.NoError(t, err)
+			if test.activeClaim {
+				require.NotNil(t, persisted.ClaimToken)
+				require.Equal(t, token, *persisted.ClaimToken)
+				require.NotNil(t, persisted.ClaimOwner)
+				require.Equal(t, "scheduler-1", *persisted.ClaimOwner)
+			} else {
+				require.Nil(t, persisted.ClaimToken)
+				require.Nil(t, persisted.ClaimOwner)
+			}
+
 			roundTripped, err := EventActionExecutionFrom(persisted)
 			require.NoError(t, err)
+
 			require.Equal(t, &execution, roundTripped)
+			if !test.activeClaim {
+				require.Equal(t, uuid.Nil, roundTripped.ClaimToken)
+				require.Empty(t, roundTripped.ClaimOwner)
+			}
 		})
 	}
 }
 
+func resultPointer(result eventrule.ExecutionResult) *eventrule.ExecutionResult {
+	return &result
+}
+
 func TestEventActionExecutionFromRejectsInvalidPersistence(t *testing.T) {
 	now := time.Date(2026, 8, 21, 12, 0, 0, 0, time.UTC)
 	execution, err := eventrule.NewExecution(uuid.New(), "notify", &eventrule.NoopPlan{}, now)
 	require.NoError(t, err)
+
 	valid, err := EventActionExecutionTo(execution)
 	require.NoError(t, err)
 
@@ -68,6 +129,7 @@ func TestEventActionExecutionFromRejectsInvalidPersistence(t *testing.T) {
 			wantErr:   "does not match plan type",
 		},
 	}
+
 	for name, test := range tests {
 		t.Run(name, func(t *testing.T) {
 			var persisted *dbmodel.EventActionExecution
@@ -76,12 +138,15 @@ func TestEventActionExecutionFromRejectsInvalidPersistence(t *testing.T) {
 				persisted = &copy
 				test.mutate(persisted)
 			}
+
 			result, err := EventActionExecutionFrom(persisted)
+
 			if test.wantErr == "" {
 				require.NoError(t, err)
 				require.Equal(t, test.wantNil, result == nil)
 				return
 			}
+
 			require.ErrorIs(t, err, eventrule.ErrInvalidPersistedExecution)
 			require.ErrorContains(t, err, test.wantErr)
 			require.Nil(t, result)