@@ -234,0 +242,4 @@
+				return nil
+			})
+		}
+		if errGroupErr := errGroup.Wait(); errGroupErr != nil {

@@ -231,3 +231,1 @@
-			} else {
-				newHeight++
-			}
+			errGroup.Go(func() error {

@@ -196,2 +196,4 @@
 	cid.Encode(rawCID)
 
+	errGroup, egCtx := errgroup.WithContext(ctx)
+	errGroup.SetLimit(1024)

@@ -231,3 +232,1 @@
-			} else {
-				newHeight++
-			}
+			errGroup.Go(func() error {

@@ -128,0 +131,4 @@
+
+	ms := make([]*pilorama.Move, len(streams))
+	for i := range streams {
+		m, ok := <-streams[i]

@@ -128,0 +152,4 @@
+	for {
+		minTimeMove := &pilorama.Move{}
+		minTimeMove.Time = math.MaxInt64
+		minTimeMoveIndex := -1

@@ -128,0 +160,4 @@
+			}
+		}
+
+		if minTimeMoveIndex == -1 {

@@ -128,0 +177,4 @@
+}
+
+func (s *Service) applyOperationStream(ctx context.Context, cid cid.ID, treeID string,
+	operationStream <-chan *pilorama.Move, notApplied chan<- *pilorama.Move) {

@@ -128,0 +182,4 @@
+	const workersCount = 1024
+	errGroup.SetLimit(workersCount)
+
+	var prev *pilorama.Move

@@ -128,0 +196,4 @@
+
+		errGroup.Go(func() error {
+			if err := s.forest.TreeApply(ctx, cid, treeID, m, true); err != nil {
+				notApplied <- m

@@ -128,0 +191,4 @@
+
+		// skip already applied op
+		if prev != nil {
+			if prev.Time == m.Time {

@@ -128,0 +146,4 @@
+	var minStreamedLastHeight uint64 = math.MaxUint64
+
+	for {
+		minTimeMove := &pilorama.Move{}

@@ -140,0 +273,4 @@
+		nodeOperationStreams[i] = make(chan *pilorama.Move)
+	}
+	merged := make(chan *pilorama.Move)
+	minStreamedLastHeight := make(chan uint64)