diff --git a/flytepropeller/pkg/controller/nodes/array/handler.go b/flytepropeller/pkg/controller/nodes/array/handler.go index a772d40bb3..bb0d1458dc 100644 --- a/flytepropeller/pkg/controller/nodes/array/handler.go +++ b/flytepropeller/pkg/controller/nodes/array/handler.go @@ -418,7 +418,7 @@ 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()) + deltaDuration := deltaTimestampSeconds(startedAt.Time, subNodeStartedAt.Time, arrayNodeState.SubNodeDeltaTimestamps.ItemSize) arrayNodeState.SubNodeDeltaTimestamps.SetItem(index, deltaDuration) } } @@ -825,3 +825,22 @@ func (a *arrayNodeHandler) buildArrayNodeContext(ctx context.Context, nCtx inter return arrayNodeExecutor, arrayExecutionContext, &arrayNodeLookup, &arrayNodeLookup, &subNodeSpec, subNodeStatus, nil } + +// deltaTimestampSeconds returns the whole seconds between the ArrayNode and subNode start times, +// clamped to the range an itemSize-bit CompactArray item can hold. A subNode may start more than +// MaxDeltaTimestamp after the ArrayNode (large fan-out with low parallelism), and a retried or +// recovered ArrayNode may have a later start time than its subNodes. Storing either value as is +// panics in SetItem on every evaluation, so we trade timestamp precision for progress. +func deltaTimestampSeconds(startedAt, subNodeStartedAt time.Time, itemSize uint) uint64 { + delta := subNodeStartedAt.Sub(startedAt) + if delta <= 0 { + return 0 + } + + seconds := uint64(delta / time.Second) // #nosec G115 + if maxValue := (uint64(1) << itemSize) - 1; seconds > maxValue { + return maxValue + } + + return seconds +} diff --git a/flytepropeller/pkg/controller/nodes/array/handler_test.go b/flytepropeller/pkg/controller/nodes/array/handler_test.go index 2cd09be2b8..e23763baaa 100644 --- a/flytepropeller/pkg/controller/nodes/array/handler_test.go +++ b/flytepropeller/pkg/controller/nodes/array/handler_test.go @@ -1581,3 +1581,32 @@ func convertListToLiterals(values []int64) *idlcore.Literal { }, } } + +func TestDeltaTimestampSeconds(t *testing.T) { + // default MaxDeltaTimestamp of 3 days needs 18 bits, so the largest storable value is 2^18 - 1 + deltaTimestamps, err := bitarray.NewCompactArray(1, bitarray.Item((3 * 24 * time.Hour).Seconds())) + assert.NoError(t, err) + maxValue := uint64(1)<