diff --git a/docs/kvm2-provider.md b/docs/kvm2-provider.md index 7f41cda..251d0fc 100644 --- a/docs/kvm2-provider.md +++ b/docs/kvm2-provider.md @@ -285,6 +285,18 @@ operations share. A cancelled or overlong call kills the connection it ran on (see *Timeouts and connections*), and a poll for some other Proxy being cancelled must not cut off a 20-minute upload. +A body that fails part way through is reported to the caller **out of +band**: the reader handed to `StorageVolUpload` turns any source error +into a clean end of stream and keeps the real error for afterwards, so +the short upload is caught by the length and digest checks rather than +by the stream. This is not tidiness. go-libvirt answers a reader error +by parking its stream sender forever on a channel it no longer reads, so +handing it one wedges the call and leaks the sender with its 4 MiB +buffer; see *Known gaps*. For the same reason the download carries its +own deadline, a minute inside the transfer budget, so a stalled mirror +ends through its body rather than through the killed socket the budget +would otherwise leave behind. + The download needs a `Content-Length`: the volume's capacity is declared before the upload starts. The request sends `Accept-Encoding: identity` so a transparently gzipped body cannot hide the length. A response @@ -541,6 +553,30 @@ never contain key bytes, the known_hosts content or cloud-init. ## Known gaps +- **An upload that libvirtd itself fails leaks a goroutine, upstream.** + go-libvirt's `requestStream` hands its stream sender an unbuffered + channel and then, on the error path, returns without ever reading it: + + ```go + abort := make(chan bool) // rpc.go + outErr := make(chan error) + go func() { outErr <- l.socket.SendStream(...) }() + resp, err = l.processIncomingStream(c, in) + if err != nil { + abort <- true + return resp, err // outErr is never drained + } + ``` + + Whenever the daemon errors mid-upload — a pool that fills, a volume + removed underneath — the sender parks on `outErr` for good, taking its + 4 MiB buffer with it. Closing the socket does not help: both ends are + blocked on channel sends, not on I/O. kvm2 avoids the two cases it + controls (a failing body, and its own transfer deadline; see *The image + cache*), so what is left needs a server-side failure. The fix is two + characters upstream, `make(chan bool, 1)` and `make(chan error, 1)`; if + it ever bites hard enough, a `replace` carrying that patch is the + escape hatch. - **Create blocks**, including the first image download on each host, holding one of the three reconcile workers for its duration (LAN speed, a few hundred MB, once per image per host). Other Proxies keep diff --git a/docs/plans-executions/2026-09-17-1515-kvm2-review-fixes.md b/docs/plans-executions/2026-09-17-1515-kvm2-review-fixes.md index f7d0bb4..69b269f 100644 --- a/docs/plans-executions/2026-09-17-1515-kvm2-review-fixes.md +++ b/docs/plans-executions/2026-09-17-1515-kvm2-review-fixes.md @@ -3,11 +3,13 @@ Plan: [docs/plans/2026-09-17-1515-kvm2-review-fixes.md](../plans/2026-09-17-1515 - [x] Step 0 — Save plan into the repo - [x] Step 1 — H1: a stalled write must not delay the teardown +- [x] Step 2 — H2: a source error must not reach go-libvirt's stream sender +- [ ] Step 3 — H3: an unreadable address must not look like a missing one +- [ ] Step 4 — H4: the base image cache must not be shared across provider entries -Remaining: the other 51 findings in `review/REPORT.md`, in severity order — H2 (the -`StorageVolUpload` stream wedge), H3 (`address()` swallowing errors), H4 (the shared base -image cache), then the mediums. Each gets its own checklist line here when it is planned, -so this list stays a record of decided work rather than a wish. +Remaining after those: the 48 mediums and lows in `review/REPORT.md`. Each gets its own +checklist line here when it is planned, so this list stays a record of decided work rather +than a wish. ## Step 0 — plan committed @@ -122,3 +124,89 @@ It now states the order (mark dead, then kill) and why both steps are lock-free. **No CHANGELOG entry yet** — the repo's rule is to add one once the user confirms the change works, so it lands with a later "validate and record" step, as in the kvm campaign. + +## Step 2 — H2: a source error must not reach go-libvirt's stream sender + +This one is a workaround, not a repair. The defect is upstream, and the entry says so in +three places (code comment, `docs/kvm2-provider.md` *Known gaps*, commit message) so nobody +later reads the wrapper as ours to delete. + +`requestStream` in the pinned go-libvirt starts its stream sender on two **unbuffered** +channels and then, on the error path, returns without ever reading `outErr`. A non-EOF error +from the reader makes `SendStream` emit a StatusError packet and return, whereupon it parks on +`outErr <-` forever; libvirtd answers the StatusError with an error, so the main goroutine +parks on `abort <- true` forever. `StorageVolUpload` never returns. The 20-minute budget +rescues only the caller, and the socket kill does nothing at all here, because both ends are +blocked on channel sends rather than on I/O — which is what the comment at `backend.go:152` +got wrong. Every failed upload leaked two goroutines and 4 MiB against a 128Mi pod limit, +with the reconciler retrying on backoff. + +Three changes. + +`eofReader` sits at the go-libvirt seam in `backend.go`, not in `image.go`, so it also covers +the seed and marker uploads, and so it stays out of the file Step 4 rewrites. It converts any +source error to `io.EOF` — the one outcome that ends a stream the normal way — and keeps the +real error for `StorageVolUpload` to return afterwards. `countingReader` is untouched +underneath it and still sees the true error, so `uploadImage`'s existing length and digest +checks produce exactly the messages they did before. + +The download now carries its own deadline, one minute inside the transfer budget. +`fetchImage` already bounded the whole fetch at 20 minutes, so the body's death and the +budget's socket kill were racing, and the socket usually won. With the body a minute ahead, a +stalled mirror ends through the reader and unwinds cleanly. This is the difference between a +slow mirror costing a Transient error and costing a permanent leak, and slow mirrors are the +realistic repeat offender. + +`discardVolume` moved from `uploadImage` up into `fetchImage`'s upload-error branch, so the +dedicated connection is closed — and libvirtd has let go of the stream's fd — before the file +is unlinked. Moving it rather than adding a second `Close` is what keeps +`image_test.go`'s `closeCount() == 1` assertion true. `uploadImage` lost its `be` parameter +and its private logger in the process. + +**What is left, deliberately.** A failure that originates in libvirtd mid-upload (ENOSPC, or +the volume deleted underneath) still leaks, because by then `SendStream` has already returned +and nothing we pass, cancel or close can reach it. Confirmed unfixable from this side. The +upstream fix is two characters and is quoted in *Known gaps* along with the `replace` escape +hatch, should it ever bite hard enough to be worth the maintenance. + +**Unverified, ~85%.** That libvirtd answers a short-but-clean stream finish with OK rather +than an error. `virFDStreamWrite` enforces `length` only on writes past it and nothing +compares `offset` to `length` on the close path, which is also why `virsh vol-upload` of a +small file into a large volume works. If it answers with an error we land back in a deadlock +by a different door — but that is precisely today's behaviour, so the change cannot make +things worse. To settle it, on any box with a libvirtd (not beelink, which is read-only +here), against the pinned go-libvirt: + +```sh +virsh -c qemu:///session pool-define-as probe dir --target ~/vmprobe +virsh -c qemu:///session pool-start probe +virsh -c qemu:///session vol-create-as probe short.raw 10M --format raw +# then a ~40-line main.go: StorageVolUpload(vol, r, 0, 10<<20, 0) where r +# yields 1 MiB then io.EOF; print the error, then runtime.NumGoroutine() +# and a goroutine profile after a two-second settle. +``` + +Expected: `err == nil` and the goroutine count back at baseline. The same program with a +reader that returns a non-EOF error should wedge against the *unpatched* seam, which also +confirms the diagnosis end to end. + +The negative check that matters was run on the unit test instead — `eofReader.Read` was +changed back to returning the real error, and both new tests failed with +`io.Copy through eofReader = connection reset by peer, want nil`, then passed again once +restored. + +```bash +go build ./... && go vet ./... && go test -race -count=1 ./internal/provider/kvm2/ +``` + +### Worth noting + +The `fetchImage` discard is still unfenced against a second provider entry that shares the +volume name — Step 4 (H4) is what closes that, by making the name per-entry. The deferred +atomic-publish work then lands exactly where this step put the discard. + +**Deliberately not tested:** the abort protocol end to end. `libvirttest` reads a fixed +28-byte header per packet and implements no stream procedures, so pinning this hermetically +would mean a 150-250 line libvirt RPC server stub whose only assertion is about *upstream's* +goroutines. Same call as H1's: the test would cost more than it protects. The two tests that +did land pin the one property the seam actually needs. diff --git a/internal/provider/kvm2/backend.go b/internal/provider/kvm2/backend.go index 22fa1e5..648e8c9 100644 --- a/internal/provider/kvm2/backend.go +++ b/internal/provider/kvm2/backend.go @@ -2,6 +2,7 @@ package kvm2 import ( "context" + "errors" "fmt" "io" "sync" @@ -90,6 +91,12 @@ const smallUploadBytes = 16 << 20 // closeBudget bounds the courtesy ConnectClose RPC in Close. const closeBudget = 5 * time.Second +// transferGrace keeps the image download's own deadline inside the transfer +// budget. A stalled mirror then ends by its body failing, which the upload +// seam turns into a clean end of stream, instead of by the budget killing +// the socket, which go-libvirt's stream sender does not survive. +const transferGrace = time.Minute + func (b budgets) upload(length uint64) time.Duration { if length <= smallUploadBytes { return b.seedUpload @@ -325,9 +332,50 @@ func (b *libvirtBackend) StorageVolCreateXML(ctx context.Context, pool libvirt.S } func (b *libvirtBackend) StorageVolUpload(ctx context.Context, vol libvirt.StorageVol, r io.Reader, offset, length uint64, flags libvirt.StorageVolUploadFlags) error { - return b.link.run(ctx, "StorageVolUpload", b.budgets.upload(length), func() error { - return b.lv.StorageVolUpload(vol, r, offset, length, flags) + src := &eofReader{r: r} + err := b.link.run(ctx, "StorageVolUpload", b.budgets.upload(length), func() error { + return b.lv.StorageVolUpload(vol, src, offset, length, flags) }) + if err == nil { + if srcErr := src.sourceErr(); srcErr != nil { + return fmt.Errorf("reading the upload source for %s: %w", vol.Name, srcErr) + } + } + return err +} + +// eofReader hides a source error from go-libvirt's stream sender and keeps +// it for the caller. SendStream answers a non-EOF error with a StatusError +// packet and then blocks forever handing its result to an unbuffered +// channel that requestStream no longer reads (go-libvirt rpc.go), which +// wedges the call and leaks the goroutine along with its 4 MiB buffer; +// closing the socket does not help, because both ends are parked on +// channel sends rather than on I/O. io.EOF is the one outcome that ends +// the stream the normal way. The field is guarded because an abandoned +// call returns while go-libvirt is still reading. +type eofReader struct { + r io.Reader + mu sync.Mutex + err error +} + +func (e *eofReader) Read(b []byte) (int, error) { + n, err := e.r.Read(b) + if err == nil || errors.Is(err, io.EOF) { + return n, err + } + e.mu.Lock() + if e.err == nil { + e.err = err + } + e.mu.Unlock() + return n, io.EOF +} + +func (e *eofReader) sourceErr() error { + e.mu.Lock() + defer e.mu.Unlock() + return e.err } func (b *libvirtBackend) StorageVolDownload(ctx context.Context, vol libvirt.StorageVol, w io.Writer, offset, length uint64, flags libvirt.StorageVolDownloadFlags) error { diff --git a/internal/provider/kvm2/backend_test.go b/internal/provider/kvm2/backend_test.go index 8511db6..58259b5 100644 --- a/internal/provider/kvm2/backend_test.go +++ b/internal/provider/kvm2/backend_test.go @@ -3,6 +3,7 @@ package kvm2 import ( "context" "errors" + "io" "strings" "sync/atomic" "testing" @@ -213,6 +214,57 @@ func TestMarkDead_slowKillDoesNotBlockCallers(t *testing.T) { } } +// TestEOFReader_sourceErrorNeverReachesTheStream pins the one property the +// upload seam needs: go-libvirt's stream sender must never see an error +// other than io.EOF, because it answers one by parking forever on a channel +// nothing reads. The real error has to survive for the caller. +func TestEOFReader_sourceErrorNeverReachesTheStream(t *testing.T) { + t.Parallel() + boom := errors.New("connection reset by peer") + tests := []struct { + name string + reader io.Reader + wantN int + wantErr error + }{ + {name: "error after data", reader: io.MultiReader(strings.NewReader("abc"), errReader{boom}), wantN: 3, wantErr: boom}, + {name: "error with no data", reader: errReader{boom}, wantErr: boom}, + {name: "clean eof passes through", reader: strings.NewReader("abc"), wantN: 3}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + e := &eofReader{r: tc.reader} + n, err := io.Copy(io.Discard, e) + if err != nil { + t.Fatalf("io.Copy through eofReader = %v, want nil: the stream sender must only ever see io.EOF", err) + } + if n != int64(tc.wantN) { + t.Errorf("copied %d bytes, want %d", n, tc.wantN) + } + if !errors.Is(e.sourceErr(), tc.wantErr) { + t.Errorf("sourceErr() = %v, want %v", e.sourceErr(), tc.wantErr) + } + }) + } +} + +func TestEOFReader_keepsTheFirstError(t *testing.T) { + t.Parallel() + first, second := errors.New("first"), errors.New("second") + e := &eofReader{r: io.MultiReader(errReader{first}, errReader{second})} + if _, err := io.Copy(io.Discard, e); err != nil { + t.Fatalf("io.Copy through eofReader = %v, want nil", err) + } + if !errors.Is(e.sourceErr(), first) { + t.Errorf("sourceErr() = %v, want the first error", e.sourceErr()) + } +} + +type errReader struct{ err error } + +func (r errReader) Read([]byte) (int, error) { return 0, r.err } + func TestBudgets_upload(t *testing.T) { t.Parallel() tests := []struct { diff --git a/internal/provider/kvm2/image.go b/internal/provider/kvm2/image.go index 1b5c34d..6a6a695 100644 --- a/internal/provider/kvm2/image.go +++ b/internal/provider/kvm2/image.go @@ -214,7 +214,12 @@ func (p *Provider) fetchImage(ctx context.Context, host provider.KVMHost, pool l log := p.logger(ctx).WithValues("host", host.Name, "volume", name) start := time.Now() - resp, err := p.do(ctx, http.MethodGet, src.URL) + // The body runs out of time before the surrounding budget does, so a + // mirror that stalls mid-image ends the upload through its reader + // rather than through the killed socket the budget would leave. + httpCtx, cancelHTTP := context.WithTimeout(ctx, defaultBudgets.transfer-transferGrace) + defer cancelHTTP() + resp, err := p.do(httpCtx, http.MethodGet, src.URL) if err == nil { err = checkResponse(src.URL, resp) } @@ -227,8 +232,11 @@ func (p *Provider) fetchImage(ctx context.Context, host provider.KVMHost, pool l span.SetAttributes(attribute.Int64("kvm2.bytes", size)) log.V(1).Info("fetching base image into host pool", "url", src.URL, "bytes", size) - vol, digest, err := p.uploadImage(ctx, host, pool, be, src, resp.Body, size) + vol, digest, err := p.uploadImage(ctx, host, pool, src, resp.Body, size) if err != nil { + // uploadImage has returned, so its dedicated connection is closed + // and libvirtd has let go of the stream before the file goes. + p.discardVolume(ctx, log, be, pool, name) span.SetStatus(codes.Error, err.Error()) return libvirt.StorageVol{}, err } @@ -249,10 +257,10 @@ func (p *Provider) fetchImage(ctx context.Context, host provider.KVMHost, pool l // uploadImage creates the base volume and streams body into it over a // dedicated connection with a running sha256, then checks length and -// digest. On any failure the volume is discarded and the error says why: -// a body that ended early is Transient, a digest that differs from the -// configured one is Permanent. -func (p *Provider) uploadImage(ctx context.Context, host provider.KVMHost, pool libvirt.StoragePool, be backend, src imageSource, body io.Reader, size int64) (libvirt.StorageVol, string, error) { +// digest. The error says why: a body that ended early is Transient, a +// digest that differs from the configured one is Permanent. Discarding +// the volume is the caller's, once this connection is closed. +func (p *Provider) uploadImage(ctx context.Context, host provider.KVMHost, pool libvirt.StoragePool, src imageSource, body io.Reader, size int64) (libvirt.StorageVol, string, error) { name := src.volumeName() xmlDoc, err := buildVolumeXML(volumeSpec{Name: name, CapacityBytes: size, Format: src.Format}) if err != nil { @@ -289,7 +297,6 @@ func (p *Provider) uploadImage(ctx context.Context, host provider.KVMHost, pool err = permanentf("image %s: sha256 %s does not match the configured %s", src.URL, digest, strings.ToLower(src.SHA256)) } if err != nil { - p.discardVolume(ctx, p.logger(ctx).WithValues("host", host.Name, "volume", name), be, pool, name) return libvirt.StorageVol{}, "", err } return vol, digest, nil