Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 20 additions & 1 deletion flytepropeller/pkg/controller/nodes/array/handler.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
}
Expand Down Expand Up @@ -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
}
29 changes: 29 additions & 0 deletions flytepropeller/pkg/controller/nodes/array/handler_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -1570,3 +1570,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)<<deltaTimestamps.ItemSize - 1

startedAt := time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC)
tests := []struct {
name string
subNodeStartedAt time.Time
expected uint64
}{
{name: "SameTime", subNodeStartedAt: startedAt, expected: 0},
{name: "WithinRange", subNodeStartedAt: startedAt.Add(90*time.Minute + 500*time.Millisecond), expected: 5400},
{name: "LargestStorableValue", subNodeStartedAt: startedAt.Add(time.Duration(maxValue) * time.Second), expected: maxValue}, // #nosec G115
{name: "BeyondRange", subNodeStartedAt: startedAt.Add(time.Duration(267605) * time.Second), expected: maxValue},
{name: "SubNodeStartedBeforeArrayNode", subNodeStartedAt: startedAt.Add(-time.Hour), expected: 0},
}

for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
delta := deltaTimestampSeconds(startedAt, test.subNodeStartedAt, deltaTimestamps.ItemSize)
assert.Equal(t, test.expected, delta)
assert.NotPanics(t, func() { deltaTimestamps.SetItem(0, delta) })
assert.Equal(t, test.expected, deltaTimestamps.GetItem(0))
})
}
}
Loading