Stop a cancelled context from leaving corestreamd alive forever

docs/two-box-runbook.md

@@ -72,6 +72,15 @@ stream ends when core exits. Expected ≈ P1 at n=100 (single-box read: p50
 2.19 ms; the real wire should land at or under it). Run it with
 `-compress=false` too for the paired uncompressed row (17.94 ms p50).
 
+**What P2 does NOT cover.** The target shape is 600 ms close time at sac6000
+density, and P2 does not run at it: apply-load produces a ledger every ~2 s on
+this hardware, 3.3x slower than the target, so P2 proves the tap and the codec
+against real meta at a third of the cadence. Only the synthetic source (P1)
+runs at 600 ms. So the transport is measured at the target shape and real core
+meta is measured at core's natural pace — the two have never been measured
+together, and cannot be until core closes sac6000 ledgers at 600 ms. Read a
+P2 pass as "the tap is honest", not as "the target shape is proven".
+
 **Cell P3 (optional) — today's network shape.**
 Server: `-source pipe -pipe-cmd "stellar-core catchup <recent>/<count>
 --metadata-output-stream fd:3"` on a pubnet config. Real pubnet metas are
@@ -101,3 +110,22 @@ measured against.
   penalty is the unhidden wire ≈ 2.2 ms + tail).
 - If P1 exceeds ~5.5 ms: something outside the model (irq steering, placement,
   clock) — check `chronyc tracking`, `ethtool -S` coalescing, and rerun.
+- Capture `ethtool -S` allowance counters (`bw_in_allowance_exceeded`,
+  `pps_allowance_exceeded`, `conntrack_allowance_exceeded`) and `ss -ti`
+  retransmits on the ledger socket alongside the percentiles. CloudWatch
+  cannot substitute: AWS documents that these counters can show dropped
+  packets while the instance metrics show nothing, because the averaging
+  window is far coarser than a microburst — and at 600 ms the stream IS a
+  microburst, on the wire 0.53% of the time.
+
+## Sizing at the target shape
+
+600 ms close time at sac6000 density is 14.48 MiB of meta every 600 ms:
+202 Mbit/s raw, 26.7 Mbit/s compressed, 2.19 TB/day of meta. Two consequences
+worth setting before a long run, neither of them about the wire:
+
+- **The ring stores raw**, so `-retention 10000` is ~141 GiB and covers only
+  100 minutes at this cadence. Size it in wall-clock: six hours is ~509 GiB
+  and 36,000 ledgers, plus ~25 MB/s of sustained ring writes.
+- **The subscriber must absorb ~85 GiB/hour** of meta, continuously. That is a
+  much tighter constraint than anything measured here.

internal/server/pipesource.go

@@ -80,6 +80,20 @@ func (p *pipeStream) Emissions(ctx context.Context, _ ledgerbackend.Range) iter.
 		}
 		// The parent must not hold the write end open, or EOF never arrives.
 		w.Close()
+		// Nor may a blocked read outlive the context. Everything below assumes
+		// the pipe eventually EOFs, which assumes every writer eventually
+		// exits — and a grandchild that escaped the process group does not:
+		// it keeps the write end open and this read never returns, so the
+		// defer that would close r is unreachable, Run never returns, and the
+		// daemon lives on with its listener already shut. That is not
+		// hypothetical; it left a corestreamd running for two days after an
+		// ordinary SIGTERM, its stellar-core spinning at 100% of a core.
+		// os.Pipe is poller-backed, so a deadline in the past unblocks the
+		// read at once.
+		stopUnblock := context.AfterFunc(ctx, func() {
+			_ = r.SetReadDeadline(time.Now())
+		})
+		defer stopUnblock()
 		// The child is killed by CommandContext on ctx cancel; reap it
 		// exactly once on every exit path so a yield-stop cannot leak a
 		// zombie and no path ever sees a second Wait's spurious error.
@@ -99,13 +113,25 @@ func (p *pipeStream) Emissions(ctx context.Context, _ ledgerbackend.Range) iter.
 		// r.Close is then a harmless double-close.
 		defer func() {
 			_ = r.Close()
-			if cmd.Process != nil && ctx.Err() == nil {
+			if cmd.Process == nil {
+				_ = wait()
+				return
+			}
+			if ctx.Err() == nil {
 				// An early stop with the context still live gets no Cancel
 				// from CommandContext; tear the group down ourselves so no
 				// grandchild survives holding the pipe.
 				_ = syscall.Kill(-cmd.Process.Pid, syscall.SIGTERM)
 			}
 			_ = wait()
+			// Whatever is still in the group outlived both the SIGTERM and
+			// the child that led it. WaitDelay only escalates to the direct
+			// child, so a grandchild ignoring SIGTERM — an apply-load mid-run
+			// does exactly this — is left running with nothing to write to.
+			// The group exists solely for this command, so sweeping it is
+			// safe; on the ordinary path it is already empty and this is an
+			// ESRCH no-op.
+			_ = syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL)
 		}()
 
 		br := bufio.NewReaderSize(r, 1<<20)
@@ -144,7 +170,7 @@ func (p *pipeStream) Emissions(ctx context.Context, _ ledgerbackend.Range) iter.
 				yield(Emission{}, fmt.Errorf("pipe source: read ledger seq: %w", err))
 				return
 			}
-			body := io.MultiReader(bytes.NewReader(prefix[:n]), &frameTail{r: br, remaining: size - n})
+			body := io.MultiReader(bytes.NewReader(prefix[:n]), &frameTail{ctx: ctx, r: br, remaining: size - n})
 			if !yield(Emission{Seq: seq, Size: size, Body: body}, nil) {
 				return
 			}
@@ -156,6 +182,7 @@ func (p *pipeStream) Emissions(ctx context.Context, _ ledgerbackend.Range) iter.
 // EOF before the marker-declared length into a loud ErrUnexpectedEOF — a
 // child dying mid-frame must never read as a clean, shorter ledger.
 type frameTail struct {
+	ctx       context.Context
 	r         *bufio.Reader
 	remaining int64
 }
@@ -169,6 +196,13 @@ func (f *frameTail) Read(p []byte) (int, error) {
 	}
 	n, err := f.r.Read(p)
 	f.remaining -= int64(n)
+	if err != nil && f.ctx.Err() != nil {
+		// A shutdown that interrupts a body read reports the cancellation,
+		// not the read deadline that implemented it: the consumer tells its
+		// own shutdown from a source failure by unwrapping the context error,
+		// and an i/o timeout would be logged as a failed source loop.
+		return n, f.ctx.Err()
+	}
 	if errors.Is(err, io.EOF) && f.remaining > 0 {
 		err = fmt.Errorf("pipe source: frame truncated %d bytes short: %w", f.remaining, io.ErrUnexpectedEOF)
 	}

internal/server/pipesource_test.go

@@ -9,6 +9,7 @@ import (
 	"os"
 	"path/filepath"
 	"testing"
+	"time"
 
 	"github.com/stellar/go-stellar-sdk/xdr"
 )
@@ -233,3 +234,40 @@ func TestPipeSource_TruncatedFrameSurfaces(t *testing.T) {
 	}
 	t.Fatal("no emission yielded")
 }
+
+// TestPipeSource_CancelUnblocksAnEscapedWriter pins the shutdown contract
+// against the case that actually happened: a grandchild that escapes the
+// process group keeps the pipe's write end open, so the read never EOFs. The
+// source must still return when the context is cancelled — before this was
+// fixed it did not, and the daemon stayed alive with its listener already
+// shut down and a stellar-core spinning behind it for two days.
+func TestPipeSource_CancelUnblocksAnEscapedWriter(t *testing.T) {
+	// setsid puts the sleeper in its own session, so the source's group
+	// SIGTERM cannot reach it; it inherits fd 3 and holds the pipe open. The
+	// parent shell exits immediately, so the child is gone while the writer
+	// is not — exactly the observed shape.
+	src := PipeSource("setsid sleep 60 & exit 0")
+
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		for range src.Emissions(ctx, CountedRange(1, 0)) { //nolint:revive // draining is the point
+		}
+	}()
+
+	// The read is blocked on a pipe nothing will ever write to or close.
+	time.Sleep(200 * time.Millisecond)
+	select {
+	case <-done:
+		t.Fatal("the source ended before the cancel: the fixture is not holding the pipe open")
+	default:
+	}
+
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(10 * time.Second):
+		t.Fatal("cancel did not unblock the pipe read: the source outlived its context")
+	}
+}