feat: build and run on Windows via portable file I/O (#8)

The core read/write paths used unix.Pwrite/Pread on raw fds, and
platform_other.go imported x/sys/unix, so the library failed to compile on
Windows — breaking downstream importers there.

- Replace raw-fd syscalls with *os.File.WriteAt/ReadAt (pwrite/pread on Unix,
  OVERLAPPED on Windows; same lock-free concurrent-write property).
- Route the !linux && !darwin prealloc fallback through os.File.Truncate
  (no unix import); only platform_linux/darwin keep x/sys/unix.
- The unsorted builder's anonymous temp files use unlink-while-open on Unix
  (platform_unix.go) and FILE_FLAG_DELETE_ON_CLOSE on Windows
  (platform_windows.go); the temp dir is removed in cleanup.

Build throughput unchanged within ~5% noise. Windows is compile-verified via
cross-compile but not exercised in CI (ubuntu+macos only); the temp directory
is best-effort cleaned and may leak on a Windows crash.

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: tamirms

CLAUDE.md

@@ -25,18 +25,21 @@ Three build paths: `NewSortedBuilder` (sorted input), `NewUnsortedBuilder`
 
 ## Platform
 
-Unix-only: `platform_{linux,darwin,other}.go` all import `golang.org/x/sys/unix`
-(for `fallocate`/`fcntl`), so the library does **not** build on Windows. CI runs
-on ubuntu + macos only.
+`platform_linux.go`/`platform_darwin.go` use `golang.org/x/sys/unix` (for
+`fallocate`/`fcntl`); `platform_other.go` (`!linux && !darwin`, incl. Windows)
+uses `os.File.Truncate`. File I/O is portable `*os.File.WriteAt`/`ReadAt`, and
+the unsorted builder's temp files use unlink-while-open on Unix and
+`FILE_FLAG_DELETE_ON_CLOSE` on Windows. Builds and runs on Windows, but CI is
+ubuntu + macos only — so Windows is compile-verified, not runtime-tested.
 
 ## Concurrency
 
 The builders are heavily concurrent and are the main correctness risk:
 - **Sorted parallel** (`builder_parallel.go`): an `errgroup` worker pool builds
   blocks; a single writer goroutine emits them in block order via channels.
 - **Unsorted** (`builder_unsorted*.go`): concurrent writers each own a temp file
-  (lock-free `pwrite`); the finish phase uses a reader goroutine pool with
-  per-slot fences.
+  (lock-free positional `WriteAt`); the finish phase uses a reader goroutine pool
+  with per-slot fences.
 
 Treat the race detector as a first-class gate, and keep per-test timeouts on the
 channel/`errgroup` code so a deadlock fails the run instead of hanging it. If a

builder_unsorted.go

@@ -8,12 +8,11 @@ import (
 	"math"
 	"math/bits"
 	"os"
+	"path/filepath"
 	"sync"
 	"sync/atomic"
 	"unsafe"
 
-	"golang.org/x/sys/unix"
-
 	intbits "github.com/stellar/streamhash/internal/bits"
 	"github.com/stellar/streamhash/internal/sherr"
 )
@@ -124,7 +123,7 @@ type unsortedBuffer struct {
 	numWriters    int        // total writer count (from config)
 	regionSize    int64      // bytes per partition region per writer file
 	writerFiles   []*os.File // [numWriters]
-	writerFds     []int      // [numWriters] fd cache for pwrite/pread
+	partsDir      string     // temp dir; removed in cleanup
 	writerCursors [][]int64  // [numWriters][numPartitions] byte offset within region
 	nextWriterID  int        // next writerID to assign
 	writerMu      sync.Mutex // protects nextWriterID
@@ -206,7 +205,6 @@ func newUnsortedBuffer(cfg *buildConfig, numBlocks uint32, numWriters int) (*uns
 		numWriters:       numWriters,
 		regionSize:       regionSize,
 		writerFiles:      make([]*os.File, numWriters),
-		writerFds:        make([]int, numWriters),
 		writerCursors:    make([][]int64, numWriters),
 	}
 	for w := range numWriters {
@@ -220,8 +218,8 @@ func newUnsortedBuffer(cfg *buildConfig, numBlocks uint32, numWriters int) (*uns
 	return u, nil
 }
 
-// createWriterFiles creates the temp directory and per-writer files.
-// Each writer gets a single file pre-allocated via fallocate.
+// createWriterFiles creates the temp directory and per-writer files, each
+// pre-allocated then detached (unlinked on Unix, DELETE_ON_CLOSE on Windows).
 func (u *unsortedBuffer) createWriterFiles(tempDir string) error {
 	if tempDir == "" {
 		tempDir = os.TempDir()
@@ -230,33 +228,34 @@ func (u *unsortedBuffer) createWriterFiles(tempDir string) error {
 	if err != nil {
 		return err
 	}
+	u.partsDir = dir
 
 	fileSize := u.regionSize * int64(u.numPartitions)
 	for w := range u.numWriters {
-		path := fmt.Sprintf("%s/writer-%03d", dir, w)
-		f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE, 0600)
+		path := filepath.Join(dir, fmt.Sprintf("writer-%03d", w))
+		f, err := openWriterFile(path)
 		if err != nil {
 			u.closeWriterFiles()
 			os.RemoveAll(dir)
 			return fmt.Errorf("create writer %d: %w", w, err)
 		}
-		if err := preallocFile(int(f.Fd()), fileSize); err != nil {
+		if err := preallocFile(f, fileSize); err != nil {
 			f.Close()
 			u.closeWriterFiles()
 			os.RemoveAll(dir)
-			return fmt.Errorf("fallocate writer %d (%d bytes): %w", w, fileSize, err)
+			return fmt.Errorf("preallocate writer %d (%d bytes): %w", w, fileSize, err)
 		}
-		if err := os.Remove(path); err != nil {
+		if err := unlinkWhileOpen(path); err != nil {
 			f.Close()
 			u.closeWriterFiles()
 			os.RemoveAll(dir)
 			return fmt.Errorf("unlink writer %d: %w", w, err)
 		}
 		u.writerFiles[w] = f
-		u.writerFds[w] = int(f.Fd())
 	}
 
-	// All files are unlinked — remove the now-empty temp directory.
+	// Unix: files are unlinked, so the dir is empty and removed now. Windows:
+	// DELETE_ON_CLOSE files persist until close, so cleanup() removes the dir.
 	os.Remove(dir)
 	return nil
 }
@@ -399,9 +398,9 @@ func (ws *writerState) doFlush(bufIdx int) error {
 			return fmt.Errorf("partition %d overflow in writer %d: region capacity exceeded", p, ws.writerID)
 		}
 
-		_, err := unix.Pwrite(u.writerFds[ws.writerID], buf[:needed], off)
+		_, err := u.writerFiles[ws.writerID].WriteAt(buf[:needed], off)
 		if err != nil {
-			return fmt.Errorf("pwrite partition %d writer %d: %w", p, ws.writerID, err)
+			return fmt.Errorf("write partition %d writer %d: %w", p, ws.writerID, err)
 		}
 
 		u.writerCursors[ws.writerID][p] = cursor + int64(needed)
@@ -429,10 +428,8 @@ func (ws *writerState) flushAll() error {
 }
 
 // drainFlush waits for any in-flight background flush goroutine to finish.
-// Abort paths (Close/cleanupAll) must call this before closing the writer's fd:
-// doFlush issues unix.Pwrite on the cached fd from a background goroutine, so
-// closing the fd while that goroutine is mid-write races a closed (and possibly
-// recycled) descriptor. The normal Finish path drains via flushAll instead.
+// Abort paths (Close/cleanupAll) must drain before closing the file, or a
+// background WriteAt aborts with ErrClosed and drops its data.
 func (ws *writerState) drainFlush() {
 	ws.flushWg.Wait()
 }

builder_unsorted_finish.go

@@ -5,12 +5,12 @@ import (
 	"encoding/binary"
 	"errors"
 	"fmt"
+	"io"
 	"math/bits"
+	"os"
 	"sync"
 	"unsafe"
 
-	"golang.org/x/sys/unix"
-
 	intbits "github.com/stellar/streamhash/internal/bits"
 	"github.com/stellar/streamhash/internal/sherr"
 )
@@ -93,7 +93,7 @@ func (u *unsortedBuffer) readPartition(p int, slot *readSlot) ([][]routedEntry,
 		if regionBytes == 0 {
 			continue
 		}
-		if err := u.readRegion(u.writerFds[w], regionStart, regionBytes, readBuf, entrySize,
+		if err := u.readRegion(u.writerFiles[w], regionStart, regionBytes, readBuf, entrySize,
 			func(data []byte) {
 				k0 := binary.LittleEndian.Uint64(data[0:8])
 				k1 := binary.LittleEndian.Uint64(data[8:16])
@@ -113,21 +113,18 @@ func (u *unsortedBuffer) readPartition(p int, slot *readSlot) ([][]routedEntry,
 	return blockEntries, nil
 }
 
-// readRegion reads regionBytes from fd at offset using pread, calling fn for
-// each entry. Uses pread (not seek+read) for concurrent reader safety.
-func (u *unsortedBuffer) readRegion(fd int, offset, regionBytes int64, readBuf []byte, entrySize int, fn func([]byte)) error {
+// readRegion reads regionBytes from r at offset and invokes fn for each
+// fixed-size entry. ReadAt is positional, so readers can share one source.
+func (u *unsortedBuffer) readRegion(r io.ReaderAt, offset, regionBytes int64, readBuf []byte, entrySize int, fn func([]byte)) error {
 	pos := offset
 	remaining := regionBytes
 	leftoverN := 0
 
 	for remaining > 0 {
 		toRead := min(int64(len(readBuf)-leftoverN), remaining)
-		nr, err := unix.Pread(fd, readBuf[leftoverN:leftoverN+int(toRead)], pos)
-		if nr == 0 {
-			if err != nil {
-				return err
-			}
-			return fmt.Errorf("pread: unexpected zero read at offset %d with %d remaining", pos, remaining)
+		nr, err := r.ReadAt(readBuf[leftoverN:leftoverN+int(toRead)], pos)
+		if int64(nr) != toRead {
+			return fmt.Errorf("readRegion: short read at offset %d: got %d of %d bytes (%d remaining): %w", pos, nr, toRead, remaining, err)
 		}
 		pos += int64(nr)
 		remaining -= int64(nr)
@@ -143,10 +140,6 @@ func (u *unsortedBuffer) readRegion(fd int, offset, regionBytes int64, readBuf [
 		if len(data) > 0 {
 			leftoverN = copy(readBuf, data)
 		}
-
-		if err != nil {
-			return err
-		}
 	}
 	return nil
 }
@@ -225,8 +218,9 @@ func (u *unsortedBuffer) prepareForRead(numSlots int) error {
 	return nil
 }
 
-// cleanup closes all writer fds. The temp directory is removed during
-// createWriterFiles after all files are unlinked. Idempotent.
+// cleanup closes writer files and removes the temp dir. Best-effort and
+// idempotent: teardown errors are ignored so they can't fail a built index
+// (cleanup runs before finalize on the success path).
 func (u *unsortedBuffer) cleanup() error {
 	for w, f := range u.writerFiles {
 		if f != nil {
@@ -235,6 +229,10 @@ func (u *unsortedBuffer) cleanup() error {
 		}
 	}
 	u.writerFiles = nil
+	if u.partsDir != "" {
+		os.RemoveAll(u.partsDir)
+		u.partsDir = ""
+	}
 	return nil
 }
 

builder_unsorted_test.go

@@ -5,6 +5,7 @@ import (
 	"math/bits"
 	randv2 "math/rand/v2"
 	"os"
+	"runtime"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -132,6 +133,9 @@ func TestUnsortedBuffer_CleanupIdempotent(t *testing.T) {
 }
 
 func TestUnsortedBuffer_TempDirRemovedAfterCreate(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		t.Skip("Windows DELETE_ON_CLOSE temp files persist until cleanup, so the dir is not empty right after create")
+	}
 	tmpDir := t.TempDir()
 	cfg := &buildConfig{totalKeys: 100, payloadSize: 4, unsortedTempDir: tmpDir}
 	numBlocks, _ := numBlocksForAlgo(AlgoBijection, 100, 4, 0)

index_writer.go

@@ -8,14 +8,12 @@ import (
 	"os"
 
 	"github.com/cespare/xxhash/v2"
-	"golang.org/x/sys/unix"
 )
 
-// indexWriter handles writing index data to disk using pwrite-based writes.
+// indexWriter writes index data to disk via positional *os.File.WriteAt calls.
 // File layout: [Header 64B][UserMetaLen 4B][UserMeta][AlgoConfigLen 4B][AlgoConfig][RAM Index (N+1)×10B][Payload Region][Metadata Region][Footer 32B]
 type indexWriter struct {
 	file *os.File
-	fd   int // cached file descriptor for pwrite syscalls
 
 	// Region offsets (computed upfront for separated layout)
 	payloadRegionOffset  uint64
@@ -84,14 +82,13 @@ func newIndexWriter(path string, cfg *buildConfig, numBlocks uint32, algo blockB
 	metadataRegionOffset := payloadRegionOffset + payloadRegionSize
 
 	// Pre-allocate disk blocks for space reservation
-	if err := preallocFile(int(file.Fd()), int64(estimatedSize)); err != nil {
+	if err := preallocFile(file, int64(estimatedSize)); err != nil {
 		primaryErr := fmt.Errorf("failed to allocate disk space: %w", err)
 		return nil, errors.Join(primaryErr, file.Close())
 	}
 
 	iw := &indexWriter{
 		file:                 file,
-		fd:                   int(file.Fd()),
 		payloadRegionOffset:  payloadRegionOffset,
 		metadataRegionOffset: metadataRegionOffset,
 		userMetadataOffset:   userMetadataOffset,
@@ -136,7 +133,7 @@ func (iw *indexWriter) writePayloadsDirect(data []byte, offset uint64) error {
 	if absoluteOffset+uint64(len(data)) > iw.metadataRegionOffset {
 		return fmt.Errorf("writePayloadsDirect: write exceeds payload region boundary")
 	}
-	_, err := unix.Pwrite(iw.fd, data, int64(absoluteOffset))
+	_, err := iw.file.WriteAt(data, int64(absoluteOffset))
 	return err
 }
 
@@ -165,7 +162,7 @@ func (iw *indexWriter) recordAndWriteMetadata(metadata []byte) error {
 	})
 
 	if len(metadata) > 0 {
-		if _, err := unix.Pwrite(iw.fd, metadata, int64(iw.metadataWriteOffset)); err != nil {
+		if _, err := iw.file.WriteAt(metadata, int64(iw.metadataWriteOffset)); err != nil {
 			return err
 		}
 		if _, err := iw.metadataHasher.Write(metadata); err != nil {
@@ -202,7 +199,7 @@ func (iw *indexWriter) commitBlockWithData(metadata []byte, payloads []byte, num
 	entrySize := iw.cfg.payloadSize + iw.cfg.fingerprintSize
 	if entrySize > 0 && numKeys > 0 {
 		payloadOffset := iw.payloadRegionOffset + iw.keyCount*uint64(entrySize)
-		if _, err := unix.Pwrite(iw.fd, payloads, int64(payloadOffset)); err != nil {
+		if _, err := iw.file.WriteAt(payloads, int64(payloadOffset)); err != nil {
 			return err
 		}
 		iw.foldPayloadHash(xxhash.Sum64(payloads))
@@ -245,7 +242,7 @@ func (iw *indexWriter) finalize() error {
 	// Write header at offset 0
 	var headerBuf [headerSize]byte
 	iw.header.encodeTo(headerBuf[:])
-	if _, err := unix.Pwrite(iw.fd, headerBuf[:], 0); err != nil {
+	if _, err := iw.file.WriteAt(headerBuf[:], 0); err != nil {
 		primaryErr := fmt.Errorf("write header failed: %w", err)
 		return errors.Join(primaryErr, iw.close())
 	}
@@ -255,7 +252,7 @@ func (iw *indexWriter) finalize() error {
 	userMetaBuf := make([]byte, 4+len(iw.userMetadata))
 	binary.LittleEndian.PutUint32(userMetaBuf, uint32(len(iw.userMetadata)))
 	copy(userMetaBuf[4:], iw.userMetadata)
-	if _, err := unix.Pwrite(iw.fd, userMetaBuf, int64(iw.userMetadataOffset)); err != nil {
+	if _, err := iw.file.WriteAt(userMetaBuf, int64(iw.userMetadataOffset)); err != nil {
 		primaryErr := fmt.Errorf("write user metadata failed: %w", err)
 		return errors.Join(primaryErr, iw.close())
 	}
@@ -267,7 +264,7 @@ func (iw *indexWriter) finalize() error {
 	if algoConfigLen > 0 {
 		iw.algo.EncodeGlobalConfigInto(algoConfigBuf[4:])
 	}
-	if _, err := unix.Pwrite(iw.fd, algoConfigBuf, int64(iw.algoConfigOffset)); err != nil {
+	if _, err := iw.file.WriteAt(algoConfigBuf, int64(iw.algoConfigOffset)); err != nil {
 		primaryErr := fmt.Errorf("write algo config failed: %w", err)
 		return errors.Join(primaryErr, iw.close())
 	}
@@ -278,7 +275,7 @@ func (iw *indexWriter) finalize() error {
 		offset := i * int(ramIndexEntrySize)
 		encodeRAMIndexEntryTo(entry, ramIndexBuf[offset:])
 	}
-	if _, err := unix.Pwrite(iw.fd, ramIndexBuf, int64(iw.ramIndexOffset)); err != nil {
+	if _, err := iw.file.WriteAt(ramIndexBuf, int64(iw.ramIndexOffset)); err != nil {
 		primaryErr := fmt.Errorf("write RAM index failed: %w", err)
 		return errors.Join(primaryErr, iw.close())
 	}
@@ -290,7 +287,7 @@ func (iw *indexWriter) finalize() error {
 	}
 	var footerBuf [footerSize]byte
 	ftr.encodeTo(footerBuf[:])
-	if _, err := unix.Pwrite(iw.fd, footerBuf[:], int64(iw.metadataWriteOffset)); err != nil {
+	if _, err := iw.file.WriteAt(footerBuf[:], int64(iw.metadataWriteOffset)); err != nil {
 		primaryErr := fmt.Errorf("write footer failed: %w", err)
 		return errors.Join(primaryErr, iw.close())
 	}

platform_darwin.go

@@ -2,7 +2,11 @@
 
 package streamhash
 
-import "golang.org/x/sys/unix"
+import (
+	"os"
+
+	"golang.org/x/sys/unix"
+)
 
 // preallocFile pre-allocates disk blocks without changing the file size.
 // On macOS, uses fcntl F_PREALLOCATE to reserve contiguous blocks.
@@ -11,12 +15,12 @@ import "golang.org/x/sys/unix"
 // on sparse pre-allocated files. ftruncate would be slightly faster (~1.7μs)
 // but causes page cache pressure proportional to file size, which thrashes
 // under memory pressure.
-func preallocFile(fd int, size int64) error {
+func preallocFile(f *os.File, size int64) error {
 	fst := unix.Fstore_t{
 		Flags:   unix.F_ALLOCATEALL,
 		Posmode: unix.F_PEOFPOSMODE,
 		Offset:  0,
 		Length:  size,
 	}
-	return unix.FcntlFstore(uintptr(fd), unix.F_PREALLOCATE, &fst)
+	return unix.FcntlFstore(f.Fd(), unix.F_PREALLOCATE, &fst)
 }

platform_linux.go

@@ -2,9 +2,13 @@
 
 package streamhash
 
-import "golang.org/x/sys/unix"
+import (
+	"os"
+
+	"golang.org/x/sys/unix"
+)
 
 // preallocFile pre-allocates disk blocks and extends the file to the given size.
-func preallocFile(fd int, size int64) error {
-	return unix.Fallocate(fd, 0, 0, size)
+func preallocFile(f *os.File, size int64) error {
+	return unix.Fallocate(int(f.Fd()), 0, 0, size)
 }