Run CI on Linux, macOS and Windows; fix Tier 2 bugs

.github/workflows/go.yml

@@ -9,8 +9,12 @@ on:
 jobs:
 
   build:
-    name: Build
-    runs-on: ubuntu-latest
+    name: Build (${{ matrix.os }})
+    strategy:
+      fail-fast: false
+      matrix:
+        os: [ ubuntu-latest, macos-latest, windows-latest ]
+    runs-on: ${{ matrix.os }}
     steps:
     - name: Check out code into the Go module directory
       uses: actions/checkout@v7
@@ -24,10 +28,24 @@ jobs:
       run: go mod download
 
     - name: golangci-lint
+      if: runner.os == 'Linux'
       uses: golangci/golangci-lint-action@v9
       with:
         version: v2.12
 
     - name: Test
+      if: runner.os != 'Windows'
       run: make test
 
+    # make is not reliably available on Windows runners
+    - name: Test (Windows)
+      if: runner.os == 'Windows'
+      run: go test -timeout=5m ./...
+
+    - name: Test with race detector
+      if: runner.os == 'Linux'
+      run: make test-race
+
+    - name: Run examples
+      if: runner.os == 'Linux'
+      run: make examples

Makefile

@@ -6,12 +6,17 @@ include release.mk
 
 ALL_SOURCES := $(shell find . -type f -name '*.go')
 
-.PHONY: fmt lint test cover coverhtml examples readme
+.PHONY: fmt lint test test-race cover coverhtml examples readme
 
 test:
 	go test -timeout=60s $(shell go list ./... | grep -v "/examples") 
 	@echo "< ALL TESTS PASS >"
 
+# the race detector slows the root package to about a minute, so allow longer than test
+test-race:
+	go test -race -timeout=10m $(shell go list ./... | grep -v "/examples")
+	@echo "< ALL RACE TESTS PASS >"
+
 update-deps: go.mod
 	GOPROXY=direct go get -u ./...
 	go mod tidy
@@ -39,9 +44,9 @@ benchmark:
 
 examples:
 	@for dir in examples/*/; do \
-		if [ -f "$$dir"*.go ]; then \
+		if ls "$$dir"*.go > /dev/null 2>&1; then \
 			echo "Running example in $$dir"; \
-			(cd "$$dir" && go run *.go > /dev/null); \
+			(cd "$$dir" && go run . > /dev/null) || exit 1; \
 		fi; \
 	done
 

config.go

@@ -1,26 +1,29 @@
 package extsort
 
 // Config holds configuration settings for external sorting operations.
-// All fields have sensible defaults and can be left as zero values to use defaults.
+// Pass a nil *Config to use DefaultConfig(). In a non-nil Config, a ChunkSize or
+// NumWorkers below 1 and a negative ChanBuffSize or SortedChanBuffSize are replaced by
+// their defaults, but a zero ChanBuffSize or SortedChanBuffSize means an unbuffered channel.
+// The sorter works on its own copy, so one Config can be shared by several sorters.
 type Config struct {
 	// ChunkSize specifies the maximum number of records to store in each chunk
 	// before writing to disk. Larger chunks use more memory but reduce I/O operations.
-	// Default: 1,000,000 records. Must be > 1.
+	// Default: 1,000,000 records. Values below 1 use the default.
 	ChunkSize int
 
 	// NumWorkers controls the maximum number of goroutines used for parallel
 	// chunk sorting and merging. More workers can improve CPU utilization on multi-core systems.
-	// Default: 2 workers. Must be > 1.
+	// Default: 2 workers. Values below 1 use the default.
 	NumWorkers int
 
-	// ChanBuffSize sets the buffer size for internal channels used during chunk merging.
-	// Larger buffers can improve throughput but use more memory.
-	// Default: 1. Must be >= 0.
+	// ChanBuffSize sets how many whole chunks can wait between reading the input and
+	// sorting them. Each buffered chunk holds up to ChunkSize records in memory.
+	// Default: 1. Zero means unbuffered; negative values use the default.
 	ChanBuffSize int
 
 	// SortedChanBuffSize sets the buffer size for the output channel that delivers
 	// sorted results. Larger buffers allow more decoupling between sorting and consumption.
-	// Default: 1000. Must be >= 0.
+	// Default: 1000. Zero means unbuffered; negative values use the default.
 	SortedChanBuffSize int
 
 	// TempFilesDir specifies the directory for temporary files during sorting.
@@ -45,31 +48,32 @@ func DefaultConfig() *Config {
 	return &Config{
 		ChunkSize:          int(1e6), // 1M
 		NumWorkers:         2,
-		ChanBuffSize:       16,
+		ChanBuffSize:       1,
 		SortedChanBuffSize: 1000,
 		TempFilesDir:       "",
 	}
 }
 
-// mergeConfig validates and normalizes a Config by replacing zero/invalid values
-// with defaults. If config is nil, returns DefaultConfig().
-// This ensures all sorter instances have valid configuration values.
+// mergeConfig returns a validated and normalized copy of c, replacing invalid values
+// with defaults. If c is nil, returns DefaultConfig().
+// The caller's Config is never modified, since it may be shared by other sorters.
 func mergeConfig(c *Config) *Config {
 	d := DefaultConfig()
 	if c == nil {
 		return d
 	}
-	if c.ChunkSize < 1 {
-		c.ChunkSize = d.ChunkSize
+	merged := *c
+	if merged.ChunkSize < 1 {
+		merged.ChunkSize = d.ChunkSize
 	}
-	if c.NumWorkers < 1 {
-		c.NumWorkers = d.NumWorkers
+	if merged.NumWorkers < 1 {
+		merged.NumWorkers = d.NumWorkers
 	}
-	if c.ChanBuffSize < 0 {
-		c.ChanBuffSize = d.ChanBuffSize
+	if merged.ChanBuffSize < 0 {
+		merged.ChanBuffSize = d.ChanBuffSize
 	}
-	if c.SortedChanBuffSize < 0 {
-		c.SortedChanBuffSize = d.SortedChanBuffSize
+	if merged.SortedChanBuffSize < 0 {
+		merged.SortedChanBuffSize = d.SortedChanBuffSize
 	}
-	return c
+	return &merged
 }

diff/diff_generic.go

@@ -107,14 +107,16 @@ func (d *differ[T]) diff() (r Result, err error) {
 			}
 		}
 	}
-	// check for errors just in case
+	// check for errors just in case. Each error channel is read once: here if its
+	// stream has ended, otherwise after the stream is drained below.
+	aErrPending, bErrPending := okA, okB
 	if !okA {
-		if err = <-d.aErrChan; err != nil {
+		if err = d.readErr(d.aErrChan); err != nil {
 			return
 		}
 	}
 	if !okB {
-		if err = <-d.bErrChan; err != nil {
+		if err = d.readErr(d.bErrChan); err != nil {
 			return
 		}
 	}
@@ -132,9 +134,11 @@ func (d *differ[T]) diff() (r Result, err error) {
 			return r, d.ctx.Err()
 		}
 	}
-	// check for A errors once again
-	if err = <-d.aErrChan; err != nil {
-		return
+	// check for A errors if not read above
+	if aErrPending {
+		if err = d.readErr(d.aErrChan); err != nil {
+			return
+		}
 	}
 	// if only B has data left
 	for okB {
@@ -150,13 +154,26 @@ func (d *differ[T]) diff() (r Result, err error) {
 			return r, d.ctx.Err()
 		}
 	}
-	// check for B errors once again
-	if err = <-d.bErrChan; err != nil {
-		return
+	// check for B errors if not read above
+	if bErrPending {
+		if err = d.readErr(d.bErrChan); err != nil {
+			return
+		}
 	}
 	return
 }
 
+// readErr waits for the error from a stream whose data channel has closed.
+// It gives up when ctx is done, so an error channel that is never closed cannot hang the diff.
+func (d *differ[T]) readErr(errChan <-chan error) error {
+	select {
+	case err := <-errChan:
+		return err
+	case <-d.ctx.Done():
+		return d.ctx.Err()
+	}
+}
+
 // PrintDiff is a utility function that can be used as a ResultFunc to print
 // differences to stdout. It formats each difference with the Delta symbol
 // (< for OLD, > for NEW) followed by the item value.

diff/diff_result_chan.go

@@ -1,5 +1,7 @@
 package diff
 
+import "context"
+
 // StringChanResult holds a single diff result from a string comparison.
 // It contains both the difference type (NEW/OLD) and the actual string value.
 // This type is used with StringResultChan to enable parallel processing of diff results.
@@ -20,11 +22,24 @@ type StringChanResult struct {
 //   - chan *StringChanResult: A channel to receive diff results from
 //
 // The caller is responsible for closing the returned channel when done.
+// The returned function blocks until each result is received; use StringResultChanContext
+// to stop waiting when a context is done.
 func StringResultChan() (StringResultFunc, chan *StringChanResult) {
+	return StringResultChanContext(context.Background())
+}
+
+// StringResultChanContext is like StringResultChan, but the returned function stops waiting
+// for the receiver once ctx is done and returns ctx.Err(), which ends the diff with that error.
+// Pass the same context to the diff.
+func StringResultChanContext(ctx context.Context) (StringResultFunc, chan *StringChanResult) {
 	c := make(chan *StringChanResult, 1)
 	f := func(d Delta, s string) error {
-		c <- &StringChanResult{D: d, S: s}
-		return nil
+		select {
+		case c <- &StringChanResult{D: d, S: s}:
+			return nil
+		case <-ctx.Done():
+			return ctx.Err()
+		}
 	}
 	return f, c
 }

diff/diff_types.go

@@ -40,7 +40,7 @@ type Delta int
 const (
 	// NEW indicates an item that exists only in the second stream (B).
 	// This represents a "new" or "added" item when comparing A to B.
-	NEW = iota // +
+	NEW Delta = iota // +
 
 	// OLD indicates an item that exists only in the first stream (A).
 	// This represents an "old" or "removed" item when comparing A to B.

diff/regression_test.go

@@ -0,0 +1,108 @@
+package diff_test
+
+// Regression tests for diff hangs and the Delta constants.
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/lanrat/extsort/diff"
+)
+
+// stream returns a closed channel holding items.
+func stream(items ...string) chan string {
+	ch := make(chan string, len(items))
+	for _, s := range items {
+		ch <- s
+	}
+	close(ch)
+	return ch
+}
+
+func ignoreResult(diff.Delta, string) error { return nil }
+
+// runDiff runs diff.Strings and fails the test if it does not return in time.
+func runDiff(t *testing.T, ctx context.Context, a, b <-chan string, aErr, bErr <-chan error, f diff.StringResultFunc) (diff.Result, error) {
+	t.Helper()
+	type result struct {
+		r   diff.Result
+		err error
+	}
+	done := make(chan result, 1)
+	go func() {
+		r, err := diff.Strings(ctx, a, b, aErr, bErr, f)
+		done <- result{r, err}
+	}()
+	select {
+	case res := <-done:
+		return res.r, res.err
+	case <-time.After(5 * time.Second):
+		t.Fatal("diff did not return within 5s")
+		return diff.Result{}, nil
+	}
+}
+
+// The error channel of the stream that ended first used to be read twice, so a
+// caller that sent one value without closing the channel hung the diff.
+func TestErrChanOfShorterStreamIsReadOnce(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		a, b []string
+	}{
+		{"A ends first", []string{"a"}, []string{"a", "b", "c"}},
+		{"B ends first", []string{"a", "b", "c"}, []string{"a"}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			aErr, bErr := make(chan error, 1), make(chan error, 1)
+			aErr <- nil // one value each, never closed
+			bErr <- nil
+			r, err := runDiff(t, context.Background(), stream(tc.a...), stream(tc.b...), aErr, bErr, ignoreResult)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if r.Common != 1 || r.ExtraA+r.ExtraB != 2 {
+				t.Errorf("unexpected result %s", r.String())
+			}
+		})
+	}
+}
+
+// Reading an error channel ignored ctx, so a channel that is never sent to or closed
+// hung the diff even past the ctx deadline.
+func TestErrChanReadRespectsContext(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
+	defer cancel()
+	never := make(chan error)
+	_, err := runDiff(t, ctx, stream("a"), stream("b"), never, never, ignoreResult)
+	if !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatalf("got error %v, want %v", err, context.DeadlineExceeded)
+	}
+}
+
+// NEW and OLD used to be untyped integer constants, so they printed as 0 and 1
+// instead of using Delta's String method.
+func TestDeltaConstantsAreTyped(t *testing.T) {
+	if got := fmt.Sprint(diff.NEW, diff.OLD); got != "> <" {
+		t.Errorf("fmt.Sprint(NEW, OLD) = %q, want %q", got, "> <")
+	}
+}
+
+// The function returned by StringResultChan blocks on its send even after ctx is done,
+// so a diff whose results are no longer read hangs. StringResultChanContext stops waiting.
+func TestStringResultChanContextStopsWaiting(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	resultFunc, results := diff.StringResultChanContext(ctx)
+	defer close(results)
+	noErr := make(chan error)
+	close(noErr)
+	time.AfterFunc(50*time.Millisecond, cancel)
+	// three differences, and nobody reads the results channel
+	_, err := runDiff(t, ctx, stream("a1", "a2", "a3"), stream(), noErr, noErr, resultFunc)
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("got error %v, want %v", err, context.Canceled)
+	}
+}

error_scenarios_test.go

@@ -153,7 +153,9 @@ func TestLargeDataElements(t *testing.T) {
 	inputChan <- &largeVal{Key: 2, Data: largeString}
 	close(inputChan)
 
-	sort, outChan, errChan := extsort.New(inputChan, fromBytesForLargeVal, largeLessThan, nil)
+	// One element per chunk, so the elements are written to disk and read back
+	config := &extsort.Config{ChunkSize: 1}
+	sort, outChan, errChan := extsort.New(inputChan, fromBytesForLargeVal, largeLessThan, config)
 	sort.Sort(context.Background())
 
 	var results []*largeVal
@@ -304,7 +306,9 @@ func TestMixedTypeComparison(t *testing.T) {
 		panic("unknown type in deserialization")
 	}
 
-	sort, outChan, errChan := extsort.New(inputChan, mixedFromBytes, mixedLessFunc, nil)
+	// One element per chunk, so both types are written to disk and read back with mixedFromBytes
+	config := &extsort.Config{ChunkSize: 1}
+	sort, outChan, errChan := extsort.New(inputChan, mixedFromBytes, mixedLessFunc, config)
 	sort.Sort(context.Background())
 
 	var results []extsort.SortType