Size the meta pipe, decouple chunk size from it, and take work off the source loop

[tamirms]

The bug this fixes

A read cannot return more than the pipe holds, and the source framed one chunk per read — so chunk size was really a function of pipe size. Against the 64 KiB default, a 14.48 MiB ledger shipped as 234 chunks of 64,926 bytes instead of 58 of 262,144, quadrupling the per-chunk cost on both ends: compression frames, WebSocket frames, decode calls, scheduling, assembly.

That is exactly the "~234 chunks per ledger" the two-box run reported and attributed to Core's burst pattern. It reproduces here, and the mean chunk size names the real culprit: -chunk-size 262144 was silently a no-op on the real-Core path.

It also explains the 18 raw fallbacks seen during Core's setup burst — 4× the chunk rate is 4× the pressure on the encoder pool.

Changes

• -pipe-size on corestreamd (default 1 MiB), and fillChunk tops a chunk up across reads so chunk size no longer tracks pipe size. Real Core now produces 58 chunks of 261,941 bytes with the pipe left at the cache-friendly 64 KiB — matching the synthetic source exactly.
• store.PutParts retains the chunks the arena already holds, instead of building a second complete 14.48 MiB copy per ledger purely to hand Put one contiguous slice.
• A run-scoped hasher goroutine overlaps the ~1.4 ms ledger checksum with reading rather than running it between two pipe reads.
• benchrunner -mode localtap — the same pipe tap, same Core, no network — so the remote path's penalty is a difference against a real local baseline at its own best setting rather than against zero.

setPipeSize goes through SyscallConn, not Fd: Fd takes the file out of the runtime poller and silently disables SetReadDeadline, which is the mechanism a cancelled context uses to unblock a parked read. The first version used Fd and reintroduced the shutdown hang fixed in #4; the test written for that hang caught it immediately.

Honest accounting

This is not a latency win. Total per-ledger went 9.28 → 9.11 ms p50, against the ~2.9 ms the removed copy and hash would suggest. That work was already hidden in pipe-wait time — the loop spends most of its life blocked on Core, so filling the gaps cost nothing and emptying them frees nothing.

The case for merging is correctness and instrumentation: -chunk-size now means what it says, 14.5 MiB less copying and allocation per ledger, and a local baseline that can actually be measured against.

🤖 Generated with Claude Code

https://claude.ai/code/session_013zyTXU8wkocJgN6mBafnou

[Copilot]

Pull request overview

This PR decouples chunk size from pipe capacity, reduces ledger copying, overlaps hashing, and adds local pipe benchmarking.

Changes:

• Adds configurable pipe sizing and multi-read chunk filling.
• Retains ledger parts and hashes asynchronously.
• Adds localtap benchmarking and platform-specific pipe support.

Reviewed changes

Copilot reviewed 11 out of 11 changed files in this pull request and generated 10 comments.

Show a summary per file

File	Final review comments
internal/store/store.go	Nit (2 votes): separate the PutParts documentation so Put remains correctly documented.
internal/server/server.go	Moderate (2 votes): clear reused parts entries to avoid retaining prior ledger arenas.
internal/server/pipesource.go	Nits (4 and 2 votes): update comments to explain pipe backpressure separately from chunk sizing.
internal/server/pipesource_test.go	Critical (4 votes): make the pipe-size assertion conditional on Linux or the supported implementation.
internal/server/pipesize_other.go	No final comments.
internal/server/pipesize_linux.go	Nit (2 votes): update the outdated comment about pipe capacity determining chunk count.
internal/server/ledgerwork.go	No final comments.
internal/server/ledgerwork_test.go	No final comments.
go.mod	No final comments.
cmd/corestreamd/main.go	Nit (3 votes): revise the -pipe-size help text to describe buffering/read capacity.
cmd/benchrunner/main.go	Moderate (2 votes): do not count local reads as protocol chunks. Moderate (2 votes): reject negative pipe capacities. Nit (2 votes): document the new localtap mode and flags.

Suppressed comments (6)

cmd/benchrunner/main.go:435

• This mode records hasEmit and hasPipeline, but collector.summary only renders those metrics inside the len(deliveries) > 0 branch. Consequently localtap reports "no emit stamps" and hides the t_emit/pipeline measurements it was added to provide. Update the summary path to render these source-local metrics even when delivery is intentionally absent.

			emit: done.Sub(start), hasEmit: true,
			pipeline: done.Sub(start), hasPipeline: true,

cmd/benchrunner/main.go:146

• The new localtap mode is accepted here, but the default validation error below still says the mode must be loopback, remote or local. An invalid mode now produces an error that omits a valid option; include localtap in that message.

	case "localtap":
		if o.pipeCmd == "" {
			return o, errors.New("--pipe-cmd is required in localtap mode")

cmd/corestreamd/main.go:97

• Negative values are accepted by this new flag, but PipeSource passes them through as a nonzero request and setPipeSize treats want <= 0 as “do not resize”; the daemon therefore silently runs with the kernel's smaller default instead of rejecting an invalid capacity. Validate pipeBytes < 0 during flag parsing (while preserving the documented zero/default behavior).

	fs.IntVar(&o.pipeBytes, "pipe-size", server.DefaultPipeBytes,
		"kernel pipe capacity for -source pipe, in bytes; bounds how much one read returns and so how large a chunk gets")

internal/server/pipesize_linux.go:41

• A negative pipeBytes is silently treated as "do not resize" because F_SETPIPE_SZ is only called when want > 0; both new CLI parsers accept --pipe-size -1 and then run with an unspecified capacity. Reject negative values before starting the source (while retaining the documented best-effort handling for positive kernel refusals) so an invalid benchmark setting cannot silently change the measurement.

	if cerr := rc.Control(func(fd uintptr) {
		if want > 0 {
			if _, e := unix.FcntlInt(fd, unix.F_SETPIPE_SZ, want); e != nil {
				opErr = e
			}
		}

internal/server/pipesource.go:80

• A negative --pipe-size currently parses successfully and reaches this branch; setPipeSize skips resizing for any non-positive request, so the command silently uses the OS default instead of the requested capacity. Reject negative values here (zero can remain the default sentinel) so invalid configuration is reported rather than ignored.

		want := p.pipeBytes
		if want == 0 {
			want = DefaultPipeBytes
		}
		if _, err := setPipeSize(r, want); err != nil {

internal/store/store.go:179

• The server now uses PutParts, but the store tests only call Put, which wraps a single slice and never exercises this new multi-part write loop. Add a store test that writes several parts, reopens the store, and verifies Get returns their exact concatenation; otherwise ordering/partial-piece regressions can pass unnoticed.

	for _, part := range parts {
		if _, werr := f.Write(part); werr != nil {
			f.Close()
			return fmt.Errorf("store: write ledger %d: %w", seq, werr)
		}

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.