cre-1601: deterministic time; f check for round; improved time median; improved workflows dedup; improved comments;

Author: mchain0

pkg/workflows/ring/plugin.go

@@ -124,22 +124,8 @@ func (p *Plugin) ObservationQuorum(_ context.Context, _ ocr3types.OutcomeContext
 	return quorumhelper.ObservationCountReachesObservationQuorum(quorumhelper.QuorumTwoFPlusOne, p.config.N, p.config.F, aos), nil
 }
 
-func (p *Plugin) Outcome(_ context.Context, outctx ocr3types.OutcomeContext, _ types.Query, aos []types.AttributedObservation) (ocr3types.Outcome, error) {
-	// Bootstrap with minimum shards on first round; subsequent rounds build on prior outcome
-	prior := &pb.Outcome{}
-	if outctx.PreviousOutcome == nil {
-		prior.Routes = map[string]*pb.WorkflowRoute{}
-		prior.State = &pb.RoutingState{Id: outctx.SeqNr, State: &pb.RoutingState_RoutableShards{RoutableShards: p.minShardCount}}
-	} else if err := proto.Unmarshal(outctx.PreviousOutcome, prior); err != nil {
-		return nil, err
-	}
-
-	currentShardHealth := make(map[uint32]int)
-	totalObservations := len(aos)
-	allWorkflows := make([]string, 0)
-	nows := make([]time.Time, 0)
-
-	// Collect shard health observations and workflows
+func (p *Plugin) collectShardInfo(aos []types.AttributedObservation) (shardHealth map[uint32]int, workflows []string, timestamps []time.Time) {
+	shardHealth = make(map[uint32]int)
 	for _, ao := range aos {
 		observation := &pb.Observation{}
 		if err := proto.Unmarshal(ao.Observation, observation); err != nil {
@@ -149,36 +135,42 @@ func (p *Plugin) Outcome(_ context.Context, outctx ocr3types.OutcomeContext, _ t
 
 		for shardID, healthy := range observation.ShardHealthStatus {
 			if healthy {
-				currentShardHealth[shardID]++
+				shardHealth[shardID]++
 			}
 		}
 
-		// Collect workflow IDs
-		allWorkflows = append(allWorkflows, observation.WorkflowIds...)
+		workflows = append(workflows, observation.WorkflowIds...)
 
-		// Collect timestamps
 		if observation.Now != nil {
-			nows = append(nows, observation.Now.AsTime())
+			timestamps = append(timestamps, observation.Now.AsTime())
 		}
 	}
+	return shardHealth, workflows, timestamps
+}
 
-	// Calculate median time
-	now := time.Now()
-	if len(nows) > 0 {
-		slices.SortFunc(nows, time.Time.Compare)
-		now = nows[len(nows)/2]
+func (p *Plugin) Outcome(_ context.Context, outctx ocr3types.OutcomeContext, _ types.Query, aos []types.AttributedObservation) (ocr3types.Outcome, error) {
+	// Bootstrap with minimum shards on first round; subsequent rounds build on prior outcome
+	prior := &pb.Outcome{}
+	if outctx.PreviousOutcome == nil {
+		prior.Routes = map[string]*pb.WorkflowRoute{}
+		prior.State = &pb.RoutingState{Id: outctx.SeqNr, State: &pb.RoutingState_RoutableShards{RoutableShards: p.minShardCount}}
+	} else if err := proto.Unmarshal(outctx.PreviousOutcome, prior); err != nil {
+		return nil, err
 	}
 
-	// Deduplicate workflows
-	workflowMap := make(map[string]bool)
-	for _, wf := range allWorkflows {
-		workflowMap[wf] = true
-	}
-	allWorkflows = make([]string, 0, len(workflowMap))
-	for wf := range workflowMap {
-		allWorkflows = append(allWorkflows, wf)
+	totalObservations := len(aos)
+	currentShardHealth, allWorkflows, nows := p.collectShardInfo(aos)
+
+	// Need at least F+1 timestamps; fewer means >F faulty nodes and we can't trust this round
+	if len(nows) < p.config.F+1 {
+		return nil, errors.New("insufficient observation timestamps")
 	}
-	slices.Sort(allWorkflows) // Ensure deterministic order
+	slices.SortFunc(nows, time.Time.Compare)
+
+	// Use the median timestamp to determine the current time
+	now := nows[len(nows)/2]
+
+	allWorkflows = uniqueSorted(allWorkflows)
 
 	// Determine desired shard count based on observations
 	healthyShardCount := uint32(0)

pkg/workflows/ring/utils.go

@@ -0,0 +1,10 @@
+package ring
+
+import "slices"
+
+func uniqueSorted(s []string) []string {
+	result := slices.Clone(s)
+	slices.Sort(result)
+	return slices.Compact(result)
+}
+

pkg/workflows/ring/utils_test.go

@@ -0,0 +1,12 @@
+package ring
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestUniqueSorted(t *testing.T) {
+	got := uniqueSorted([]string{"c", "a", "b", "a", "c"})
+	require.Equal(t, []string{"a", "b", "c"}, got)
+}