fix(propeller): clamp ArrayNode subNode delta timestamps instead of panicking

flytepropeller/pkg/controller/nodes/array/handler.go

@@ -418,8 +418,13 @@ func (a *arrayNodeHandler) Handle(ctx context.Context, nCtx interfaces.NodeExecu
 					arrayNodeState.SubNodeDeltaTimestamps.SetItem(index, 0)
 				} else if startedAt != nil && arrayNodeState.SubNodeDeltaTimestamps.GetItem(index) == 0 {
 					// otherwise if `SubNodeDeltaTimestamps` is unset, we compute the delta and set it
-					deltaDuration := uint64(subNodeStartedAt.Time.Sub(startedAt.Time).Seconds())
-					arrayNodeState.SubNodeDeltaTimestamps.SetItem(index, deltaDuration)
+					delta := subNodeStartedAt.Time.Sub(startedAt.Time)
+					maxDelta := arrayNodeState.SubNodeDeltaTimestamps.MaxValue()
+					deltaSeconds := clampedDeltaSeconds(delta, maxDelta)
+					if delta > 0 && deltaSeconds < bitarray.Item(delta/time.Second) {
+						logger.Warnf(ctx, "subNode [%d] started %v after the ArrayNode, exceeding array-node.max-delta-timestamp; storing the offset as %ds", index, delta, maxDelta)
+					}
+					arrayNodeState.SubNodeDeltaTimestamps.SetItem(index, deltaSeconds)
 				}
 			}
 
@@ -825,3 +830,14 @@ func (a *arrayNodeHandler) buildArrayNodeContext(ctx context.Context, nCtx inter
 
 	return arrayNodeExecutor, arrayExecutionContext, &arrayNodeLookup, &arrayNodeLookup, &subNodeSpec, subNodeStatus, nil
 }
+
+// clampedDeltaSeconds converts the offset between an ArrayNode's start and a subNode's start into
+// whole seconds that fit within a SubNodeDeltaTimestamps slot. Negative offsets map to 0 (unset) and
+// offsets past maxDelta saturate, so a subNode that starts long after its parent only loses start
+// precision instead of failing the round.
+func clampedDeltaSeconds(delta time.Duration, maxDelta bitarray.Item) bitarray.Item {
+	if delta <= 0 {
+		return 0
+	}
+	return min(bitarray.Item(delta/time.Second), maxDelta)
+}

flytepropeller/pkg/controller/nodes/array/handler_test.go

@@ -1570,3 +1570,24 @@ func convertListToLiterals(values []int64) *idlcore.Literal {
 		},
 	}
 }
+
+func TestClampedDeltaSeconds(t *testing.T) {
+	const maxDelta = bitarray.Item(3 * 24 * 60 * 60)
+	tests := []struct {
+		name     string
+		delta    time.Duration
+		expected bitarray.Item
+	}{
+		{"negative", -time.Minute, 0},
+		{"zero", 0, 0},
+		{"sub-second", 900 * time.Millisecond, 0},
+		{"in-range", 90*time.Minute + 500*time.Millisecond, 5400},
+		{"at-max", time.Duration(maxDelta) * time.Second, maxDelta},
+		{"past-max", 262374 * time.Second, maxDelta},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			assert.Equal(t, test.expected, clampedDeltaSeconds(test.delta, maxDelta))
+		})
+	}
+}

flytestdlib/bitarray/compact_array.go

@@ -28,9 +28,13 @@ func (a *CompactArray) validateIndex(index int) {
 	}
 }
 
+// MaxValue returns the largest Item that fits in one slot of the array.
+func (a *CompactArray) MaxValue() Item {
+	return (Item(1) << a.ItemSize) - 1
+}
+
 func (a *CompactArray) validateValue(value Item) {
-	maxValue := (uint64(1) << a.ItemSize) - 1
-	if value > maxValue {
+	if maxValue := a.MaxValue(); value > maxValue {
 		panic(fmt.Sprintf("Value [%v] is too big. Max value is [%v].", value, maxValue))
 	}
 }

flytestdlib/bitarray/compact_array_test.go

@@ -155,3 +155,11 @@ func TestBitsNeededCalculation(t *testing.T) {
 	assert.Equal(t, uint(4), bitsNeeded(8))
 
 }
+
+func TestCompactArray_MaxValue(t *testing.T) {
+	itemArray, err := NewCompactArray(5, 3)
+	assert.NoError(t, err)
+	assert.Equal(t, Item(3), itemArray.MaxValue())
+	assert.NotPanics(t, func() { itemArray.SetItem(0, itemArray.MaxValue()) })
+	assert.Panics(t, func() { itemArray.SetItem(0, itemArray.MaxValue()+1) })
+}