feat: parallel v2 phase 2 — Ordered() via batch-aligned re-sequencing

export.go

@@ -83,12 +83,19 @@ type Streamer[T any] interface {
 	// re-iterable snapshot stream; ctx and parallelSize carry over.
 	Execute() Streamer[T]
 
-	// Parallel sets worker-pool concurrency for subsequent Filter/Map/Peek
+	// Parallel sets worker-pool concurrency for a section of stateless
 	// operations: n <= 0 keeps synchronous execution, n >= 1 runs n
-	// workers. Order is not preserved for n > 1, and Distinct/Sort/
-	// Reverse/Convert/FlatMap ignore it. Each parallel-aware operation
-	// spawns its own pool.
+	// workers on the section's fused stages (Filter/Map/Peek chain into
+	// one pool). A mid-chain call closes the current section and opens a
+	// new one; stateful ops and type changes close sections too. Order is
+	// not preserved unless Ordered() follows. See
+	// docs/proposals/parallel-v2.md.
 	Parallel(int) Streamer[T]
+	// Ordered marks the current (or next) parallel section as
+	// order-preserving: elements are index-tagged and re-sequenced at the
+	// consumer, so output matches serial execution order. No-op in serial
+	// mode. Costs one index stamp and slot lookup per element.
+	Ordered() Streamer[T]
 
 	// terminal operate
 

parallel_v2_test.go

@@ -1,6 +1,8 @@
 package stream_test
 
 import (
+	"context"
+	"math/rand/v2"
 	"slices"
 	"sync"
 	"testing"
@@ -140,3 +142,130 @@ func TestParallelV2_ShortCircuitStillLean(t *testing.T) {
 		t.Fatal("fused short-circuit hung")
 	}
 }
+
+func TestParallelV2_OrderedMatchesSerial(t *testing.T) {
+	// A3 property: Ordered() output must equal serial output element-for-element.
+	// Random pipelines over a fixed multiset, with heterogeneous stage costs
+	// (sleep jitter) to force real out-of-order completion.
+	rng := rand.New(rand.NewPCG(42, 2026))
+	for trial := range 30 {
+		n := 50 + rng.IntN(300)
+		serialSrc := make([]int, n)
+		for i := range serialSrc {
+			serialSrc[i] = i % 17
+		}
+
+		seed := rng.Uint64()
+		jitter := func(v int) int {
+			r := rand.New(rand.NewPCG(seed, uint64(v)+1))
+			time.Sleep(time.Duration(r.IntN(300)) * time.Microsecond)
+			return v * 2 // observable transform
+		}
+
+		serial := stream.SliceOf(serialSrc...).
+			Filter(func(v int) bool { return v%3 != 0 }).
+			Map(jitter).
+			ToSlice()
+
+		ordered := stream.SliceOf(serialSrc...).Parallel(4).Ordered().
+			Filter(func(v int) bool { return v%3 != 0 }).
+			Map(jitter).
+			ToSlice()
+
+		if !slices.Equal(serial, ordered) {
+			t.Fatalf("trial %d: ordered diverged from serial\nserial : %v\nordered: %v", trial, serial, ordered)
+		}
+	}
+}
+
+func TestParallelV2_OrderedVsUnordered(t *testing.T) {
+	// unordered may permute (or coincide); ordered must not — smoke test
+	// with visible divergence under sleep jitter, then verify Ordered is
+	// exactly sorted-back-to-input-order after a non-reordering pipeline.
+	src := make([]int, 500)
+	for i := range src {
+		src[i] = i
+	}
+	got := stream.SliceOf(src...).Parallel(8).Ordered().
+		Map(func(v int) int { time.Sleep(time.Duration(v%7) * time.Millisecond); return v }).
+		ToSlice()
+	if !slices.Equal(got, src) {
+		t.Fatal("Ordered() must reproduce input order exactly for identity pipeline")
+	}
+}
+
+func TestParallelV2_OrderedShortCircuitAndCancel(t *testing.T) {
+	// Ordered section + Limit short-circuit: must not hang, must respect order
+	done := make(chan []int, 1)
+	go func() {
+		done <- stream.SliceOf(make([]int, 100000)...).Parallel(4).Ordered().
+			Map(func(v int) int { return v + 1 }).
+			Limit(5).ToSlice()
+	}()
+	select {
+	case got := <-done:
+		if len(got) != 5 {
+			t.Fatalf("expected 5, got %d", len(got))
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatal("ordered short-circuit hung")
+	}
+
+	// pre-cancelled ctx: ordered section yields nothing
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	if got := stream.SliceOf(1, 2, 3).WithContext(ctx).Parallel(2).Ordered().
+		Map(func(v int) int { return v }).ToSlice(); len(got) != 0 {
+		t.Fatalf("cancelled ordered section must yield nothing, got %v", got)
+	}
+}
+
+func TestParallelV2_OrderedAdversarialBatchHeads(t *testing.T) {
+	// Filter drops EVERY batch head (idx%64==0): batches must still
+	// re-sequenence correctly because filtered elements travel as holes,
+	// keeping batch boundaries and lengths intact.
+	src := make([]int, 1000)
+	for i := range src {
+		src[i] = i
+	}
+	done := make(chan []int, 1)
+	go func() {
+		done <- stream.SliceOf(src...).Parallel(4).Ordered().
+			Filter(func(v int) bool { return v%64 != 0 }).
+			Map(func(v int) int { return v }).
+			ToSlice()
+	}()
+	select {
+	case got := <-done:
+		if len(got) != 1000-16 {
+			t.Fatalf("expected %d elements, got %d", 1000-16, len(got))
+		}
+		for i, v := range got {
+			want := i + i/63 + 1 // original order minus multiples of 64
+			if v != want {
+				t.Fatalf("pos %d: got %d want %d", i, v, want)
+			}
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatal("adversarial ordered pipeline hung")
+	}
+}
+
+func TestParallelV2_OrderedAllFiltered(t *testing.T) {
+	// every element filtered: all-hole batches must flow through without
+	// hanging and yield nothing
+	done := make(chan []int, 1)
+	go func() {
+		done <- stream.SliceOf(make([]int, 1000)...).Parallel(4).Ordered().
+			Filter(func(int) bool { return false }).
+			ToSlice()
+	}()
+	select {
+	case got := <-done:
+		if len(got) != 0 {
+			t.Fatalf("expected empty, got %v", got)
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatal("all-hole ordered pipeline hung")
+	}
+}