xmin should not waitForNewRows

PeerDB-io · Mar 5, 2024 · e3f9ff9 · e3f9ff9
1 parent d659494
commit e3f9ff9
Showing 1 changed file with 8 additions and 2 deletions.
diff --git a/flow/workflows/xmin_flow.go b/flow/workflows/xmin_flow.go
@@ -101,10 +101,16 @@ func XminFlowWorkflow(
 		Range:       &protos.PartitionRange{Range: &protos.PartitionRange_IntRange{IntRange: &protos.IntPartitionRange{Start: lastPartition}}},
 	}
 
-	err = q.waitForNewRows(ctx, signalChan, state.LastPartition)
-	if err != nil {
+	if err := ctx.Err(); err != nil {
 		return err
 	}
+	for {
+		val, ok := signalChan.ReceiveAsync()
+		if !ok {
+			break
+		}
+		q.activeSignal = model.FlowSignalHandler(q.activeSignal, val, q.logger)
+	}
 
 	logger.Info("Continuing as new workflow",
 		slog.Any("Last Partition", state.LastPartition),