From 37ee3e33ee21f2f8a0d91ff047ab45cfb09328a1 Mon Sep 17 00:00:00 2001 From: tomnewton-mulga Date: Mon, 6 Jul 2026 14:57:17 +1000 Subject: [PATCH] fix(viperblock): stop chunk-uploader goroutine leak at all VB call sites (#478) * fix(viperblock): stop chunk-uploader goroutine leak at all VB call sites viperblock.New() unconditionally starts a background chunk-uploader goroutine (30 s interval) that only VB.Close() stops. Seven call sites created short-lived or state-only VB instances without calling Close(), leaking one uploader goroutine per instance. On a single node the leaked goroutines exhaust memory (predastore RSS grew ~1.3 GB/5 min at idle post-single-suite), causing liveMemGate to reject launches and the lb suite to fail. The leaked uploaders are also the primary cause of the AEAD auth failures (mulga-be2ev): a leaked VB from prepareRootVolume holds the clone-point BlockLookup and overwrites the correct live checkpoint in S3 every 30 s, leaving SeqNums inconsistent with sealed chunk ciphertext. Fix all seven sites: - instance/service_impl.go prepareRootVolume: defer StopChunkUploader + StopWALSyncer after successful viperblock.New (mirrors the EFI path's intent; full Close is unsafe here because cloneAMIToVolume already called SaveBlockState + RemoveLocalFiles) - viperblockd.go ebs.delete handler: add StopChunkUploader alongside existing StopWALSyncer - viperblockd.go ebs.unmount handler: same - viperblockd.go shutdownVolumes: same - snapshot/service_impl.go snapshotVolume: defer both stops after New - volume/service_impl.go CreateVolume: defer both stops after New - image/service_impl.go snapshotRunningVolume: defer both stops after New - image/service_impl.go snapshotStoppedVolume: defer both stops after New Closes mulga-0zzci Closes mulga-ji5fi * fix(deps): bump viperblock to 01a7e686 (AEAD regression fix) The prior pin (6af89855) carried the snapshot-AMI AEAD regression that corrupts block storage on restore. Point at the merged viperblock dev fix. --------- Co-authored-by: joshsiv-mulga --- go.mod | 2 +- go.sum | 4 ++-- spinifex/handlers/ec2/image/service_impl.go | 8 ++++++++ spinifex/handlers/ec2/instance/service_impl.go | 4 ++++ spinifex/handlers/ec2/snapshot/service_impl.go | 5 +++++ spinifex/handlers/ec2/volume/service_impl.go | 4 ++++ spinifex/services/viperblockd/viperblockd.go | 10 +++++++--- 7 files changed, 31 insertions(+), 6 deletions(-) diff --git a/go.mod b/go.mod index 3ccdb0f11..cd6dc1dbc 100644 --- a/go.mod +++ b/go.mod @@ -19,7 +19,7 @@ require ( github.com/insomniacslk/dhcp v0.0.0-20260407060928-11b94ed970f2 github.com/klauspost/cpuid/v2 v2.3.0 github.com/mulgadc/predastore v1.11.0-dev - github.com/mulgadc/viperblock v1.10.1-0.20260703054738-6af89855f547 + github.com/mulgadc/viperblock v1.10.1-0.20260706044126-01a7e686fcae github.com/nats-io/nats-server/v2 v2.14.2 github.com/nats-io/nats.go v1.52.0 github.com/opencontainers/runtime-spec v1.3.0 diff --git a/go.sum b/go.sum index 5b7addaac..801e9ee8d 100644 --- a/go.sum +++ b/go.sum @@ -1301,8 +1301,8 @@ github.com/muesli/termenv v0.16.0 h1:S5AlUN9dENB57rsbnkPyfdGuWIlkmzJjbFf0Tf5FWUc github.com/muesli/termenv v0.16.0/go.mod h1:ZRfOIKPFDYQoDFF4Olj7/QJbW60Ol/kL1pU3VfY/Cnk= github.com/mulgadc/predastore v1.11.0-dev h1:GMkwHWEANG9Oazf5Fxk8IATDNlNLcbJ15/tJ0JrdIeE= github.com/mulgadc/predastore v1.11.0-dev/go.mod h1:j2DOdAOj/c42MRI6SkZrfh9KFbIyDDJLmf63SJ2yizw= -github.com/mulgadc/viperblock v1.10.1-0.20260703054738-6af89855f547 h1:wq8oLYxyZLly31R8FcyeFSm7f3Gr9VgbH3b3l+tK60k= -github.com/mulgadc/viperblock v1.10.1-0.20260703054738-6af89855f547/go.mod h1:2ArCxLXJjSN4Idx9g8Pdq5n1Kp7HFHs2vWMZgKJT9dI= +github.com/mulgadc/viperblock v1.10.1-0.20260706044126-01a7e686fcae h1:5tscnBmYgClla4897booPnnAGjvcuY0JiOmAOBhpjtc= +github.com/mulgadc/viperblock v1.10.1-0.20260706044126-01a7e686fcae/go.mod h1:2ArCxLXJjSN4Idx9g8Pdq5n1Kp7HFHs2vWMZgKJT9dI= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U= diff --git a/spinifex/handlers/ec2/image/service_impl.go b/spinifex/handlers/ec2/image/service_impl.go index 942ab4c33..7f8351921 100644 --- a/spinifex/handlers/ec2/image/service_impl.go +++ b/spinifex/handlers/ec2/image/service_impl.go @@ -509,6 +509,10 @@ func (s *ImageServiceImpl) snapshotRunningVolume(volumeID, snapshotID string) er slog.Error("snapshotRunningVolume: failed to create viperblock instance", "volumeId", volumeID, "err", err) return errors.New(awserrors.ErrorServerInternal) } + defer func() { + vb.StopChunkUploader() + vb.StopWALSyncer() + }() if err := vb.Backend.Init(); err != nil { slog.Error("snapshotRunningVolume: failed to init backend", "volumeId", volumeID, "err", err) @@ -578,6 +582,10 @@ func (s *ImageServiceImpl) snapshotStoppedVolume(volumeID, snapshotID string) er slog.Error("snapshotStoppedVolume: failed to create viperblock instance", "volumeId", volumeID, "err", err) return errors.New(awserrors.ErrorServerInternal) } + defer func() { + vb.StopChunkUploader() + vb.StopWALSyncer() + }() if err := vb.Backend.Init(); err != nil { slog.Error("snapshotStoppedVolume: failed to init backend", "volumeId", volumeID, "err", err) diff --git a/spinifex/handlers/ec2/instance/service_impl.go b/spinifex/handlers/ec2/instance/service_impl.go index 78c2d19d0..f1dad73c4 100644 --- a/spinifex/handlers/ec2/instance/service_impl.go +++ b/spinifex/handlers/ec2/instance/service_impl.go @@ -1389,6 +1389,10 @@ func (s *InstanceServiceImpl) prepareRootVolume(input *ec2.RunInstancesInput, im slog.Error("Failed to connect to Viperblock store", "err", err) return errors.New(awserrors.ErrorServerInternal) } + defer func() { + vb.StopChunkUploader() + vb.StopWALSyncer() + }() // Initialize the backend err = vb.Backend.Init() diff --git a/spinifex/handlers/ec2/snapshot/service_impl.go b/spinifex/handlers/ec2/snapshot/service_impl.go index 44d353f9f..97cc41df1 100644 --- a/spinifex/handlers/ec2/snapshot/service_impl.go +++ b/spinifex/handlers/ec2/snapshot/service_impl.go @@ -324,6 +324,11 @@ func (s *SnapshotServiceImpl) snapshotVolume(volumeID, snapshotID string, volume if err != nil { return fmt.Errorf("new viperblock: %w", err) } + defer func() { + vb.StopChunkUploader() + vb.StopWALSyncer() + }() + if err := vb.Backend.Init(); err != nil { return fmt.Errorf("backend init: %w", err) } diff --git a/spinifex/handlers/ec2/volume/service_impl.go b/spinifex/handlers/ec2/volume/service_impl.go index 3a241bb6e..4e9e51ce3 100644 --- a/spinifex/handlers/ec2/volume/service_impl.go +++ b/spinifex/handlers/ec2/volume/service_impl.go @@ -223,6 +223,10 @@ func (s *VolumeServiceImpl) CreateVolume(input *ec2.CreateVolumeInput, accountID slog.Error("CreateVolume failed to create viperblock instance", "err", err) return nil, errors.New(awserrors.ErrorServerInternal) } + defer func() { + vb.StopChunkUploader() + vb.StopWALSyncer() + }() vb.SetDebug(false) diff --git a/spinifex/services/viperblockd/viperblockd.go b/spinifex/services/viperblockd/viperblockd.go index e19f961d4..d3b7789ff 100644 --- a/spinifex/services/viperblockd/viperblockd.go +++ b/spinifex/services/viperblockd/viperblockd.go @@ -391,8 +391,9 @@ func launchService(cfg *Config) (err error) { slog.Error("Failed to unsubscribe config topic", "volume", ebsRequest.Volume, "err", err) } } - // Stop WAL syncer and kill nbdkit process + // Stop background goroutines and kill nbdkit process if matched.VB != nil { + matched.VB.StopChunkUploader() matched.VB.StopWALSyncer() } if err := utils.KillProcess(matched.PID); err != nil { @@ -469,9 +470,11 @@ func launchService(cfg *Config) (err error) { } } - // Clean up the VB instance's background goroutine. - // This VB is state-only (LoadState/sync) — actual I/O is in the nbdkit plugin process. + // Stop background goroutines on the state-tracking VB. + // Actual I/O is in the nbdkit plugin process; sealVolumeVB below + // opens a fresh VB and calls Close() for the proper seal. if matched.VB != nil { + matched.VB.StopChunkUploader() matched.VB.StopWALSyncer() } @@ -909,6 +912,7 @@ func launchService(cfg *Config) (err error) { func shutdownVolumes(volumes []MountedVolume, inUse func(MountedVolume) bool) { for _, volume := range volumes { if volume.VB != nil { + volume.VB.StopChunkUploader() volume.VB.StopWALSyncer() } if inUse(volume) {