gsadmin/spinifex
1
0
Fork 0
mirror of https://github.com/mulgadc/spinifex.git synced 2026-10-15 09:01:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: detach abandoned viperblock engines instead of stopping their goroutines by hand (#778)

Browse Source 
* refactor: detach abandoned viperblock engines instead of stopping their goroutines by hand

viperblockd opens an engine for a single operation in ten places and ends it
by stopping the background goroutines, never reaching Close. That leaves the
chunk GC sweeper running on an engine the caller has finished with, and leaves
the volume recorded as still held, so nothing can tell an engine that was
handed over from two engines holding one volume at once.

Replace each with vb.Detach(), which stops all three goroutines and reports
the release. One site keeps its bare StopChunkUploader: the daemon-side VB
that tracks state while the nbdkit plugin owns the data path is not finished
with the volume, and must not report otherwise.

An invariants test scans the package for a goroutine stopped outside Detach,
so the pattern cannot come back one call site at a time.

* chore: bump the viperblock pin to the release carrying Detach

Detach does not exist in the previous pin, so the detach work in this branch
cannot build without it. Targeted go get of one module, no tidy — viperblock
is the only requirement that moves.
This commit is contained in:
Josh Siviour
2026-08-12 23:53:06 +10:00
committed by GitHub
parent fbc94cb28c
commit f4623658e5
5 changed files with 113 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+1 -1
View File
@@ -24,7 +24,7 @@ require (
github.com/mulgadc/bluebottle v1.16.1-0.20260812012137-683e88e465e3
github.com/mulgadc/northstar v1.16.1-0.20260812012604-5d7a622ad059
github.com/mulgadc/predastore v1.16.1-0.20260812012531-12c90bd2e14a
github.com/mulgadc/viperblock v1.16.1-0.20260812033921-fc825c93eff6
github.com/mulgadc/viperblock v1.16.1-0.20260812124023-7e0df2323c75
github.com/nats-io/nats-server/v2 v2.14.4
github.com/nats-io/nats.go v1.52.0
github.com/opencontainers/runtime-spec v1.3.0
go.sum
+2
View File
@@ -1329,6 +1329,8 @@ github.com/mulgadc/viperblock v1.16.1-0.20260812012547-bb1225320f0c h1:5PCU4696G
github.com/mulgadc/viperblock v1.16.1-0.20260812012547-bb1225320f0c/go.mod h1:8NT5u+udY1CsoKz/JAPY2SSMcQ98Omi+42r8BkoA4FY=
github.com/mulgadc/viperblock v1.16.1-0.20260812033921-fc825c93eff6 h1:QfdcaqnrDDzsO4neJNX7BdL8vzyga/WPn4A/04QZcVw=
github.com/mulgadc/viperblock v1.16.1-0.20260812033921-fc825c93eff6/go.mod h1:8NT5u+udY1CsoKz/JAPY2SSMcQ98Omi+42r8BkoA4FY=
github.com/mulgadc/viperblock v1.16.1-0.20260812124023-7e0df2323c75 h1:eqB0VOuFPldBHuQ7ELYZHSRxq+3MXbMMi9D5bfVUwe8=
github.com/mulgadc/viperblock v1.16.1-0.20260812124023-7e0df2323c75/go.mod h1:8NT5u+udY1CsoKz/JAPY2SSMcQ98Omi+42r8BkoA4FY=
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=
spinifex/services/viperblockd/detach_engines_test.go
+98
View File
@@ -0,0 +1,98 @@
package viperblockd
//test:in-package — covers the unexported shutdownVolumes and reuses the
// in-package doubles captureLogs/createTestVBWithState/volumeOpenCount that
// the rest of this package's tests are built on.
// Tests proving a teardown that abandons an engine reports the volume
// released, pinned on the same "viperblock volume closed" record a production
// alarm would watch alongside the open.
import (
"os"
"path/filepath"
"slices"
"strings"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// volumeCloseCount is volumeOpenCount's counterpart. An open with no matching
// close leaves the volume reading as permanently held, so the two are only
// meaningful together.
func volumeCloseCount(logs, volumeID string) int {
count := 0
for line := range strings.SplitSeq(logs, "\n") {
if !strings.Contains(line, `msg="viperblock volume closed"`) {
continue
}
if slices.Contains(strings.Fields(line), "volume="+volumeID) {
count++
}
}
return count
}
// TestShutdownVolumesReleasesTheEngine covers the daemon going down with
// volumes still mounted. The engine is abandoned rather than closed, so
// without a release here every volume this daemon ever mounted stays counted
// as held for as long as the metric series lives.
func TestShutdownVolumesReleasesTheEngine(t *testing.T) {
logs := captureLogs(t)
volumeID := "vol-shutdownrelease"
vb := createTestVBWithState(t, volumeID)
require.Equal(t, 1, volumeOpenCount(logs.String(), volumeID),
"the engine must have recorded an open, or the close below has nothing to match")
shutdownVolumes([]MountedVolume{{Name: volumeID, VB: vb, PID: os.Getpid()}},
func(MountedVolume) bool { return true })
assert.Equal(t, 1, volumeCloseCount(logs.String(), volumeID),
"shutdown must report the volume released")
}
// TestNoTeardownStopsGoroutinesWithoutDetaching pins the fix. Stopping the
// background goroutines by hand looks like a teardown but reports nothing, so
// the engine count only ever counts up. Detach is the one way to end an engine
// that is not Close.
//
// The single exception is the state-tracking VB the mount path keeps alive
// while the nbdkit plugin owns the data path: it stops its uploader precisely
// because it is NOT finished with the volume.
func TestNoTeardownStopsGoroutinesWithoutDetaching(t *testing.T) {
const allowedFile = "provider_handlers.go"
const allowedContext = "This daemon-side VB tracks state only"
entries, err := os.ReadDir(".")
require.NoError(t, err)
checked := 0
for _, e := range entries {
name := e.Name()
if e.IsDir() || !strings.HasSuffix(name, ".go") || strings.HasSuffix(name, "_test.go") {
continue
}
src, err := os.ReadFile(filepath.Clean(name))
require.NoError(t, err)
checked++
lines := strings.Split(string(src), "\n")
for i, line := range lines {
if !strings.Contains(line, "StopChunkUploader()") && !strings.Contains(line, "StopWALSyncer()") {
continue
}
// Look back a few lines for the comment that documents the one
// engine which stops a goroutine without being finished.
window := strings.Join(lines[max(0, i-4):i], "\n")
if name == allowedFile && strings.Contains(window, allowedContext) {
continue
}
t.Errorf("%s:%d stops a background goroutine outside Detach: %q\n"+
"an engine torn down this way never reports its release", name, i+1, strings.TrimSpace(line))
}
}
require.Positive(t, checked, "no sources scanned — this would pass vacuously")
}
spinifex/services/viperblockd/provider_handlers.go
+6 -16
View File
@@ -497,10 +497,7 @@ func handleCreateVolume(ctx context.Context, cfg *Config, msg *nats.Msg) {
respondProvider(ctx, msg, ebsprovider.CreateVolumeResponse{Versioned: ebsprovider.NewVersioned(), Error: internalError("new viperblock: %v", err)})
return
}
defer func() {
vb.StopChunkUploader()
vb.StopWALSyncer()
}()
defer vb.Detach()
vb.SetDebug(false)
if err := vb.Backend.InitCtx(ctx); err != nil {
@@ -699,8 +696,7 @@ func handleExpandVolume(ctx context.Context, cfg *Config, msg *nats.Msg) {
return
}
defer func() {
vb.StopChunkUploader()
vb.StopWALSyncer()
vb.Detach()
cfg.releaseVolumeLease(ctx, lease)
}()
@@ -979,8 +975,7 @@ func copySnapshotOnVolume(ctx context.Context, cfg *Config, volumeID, srcSnapsho
return nil, err
}
defer func() {
vb.StopChunkUploader()
vb.StopWALSyncer()
vb.Detach()
cfg.releaseVolumeLease(ctx, lease)
}()
return copySnapshotMetaWithVB(vb, volumeID, srcSnapshotID, dstSnapshotID)
@@ -1162,10 +1157,7 @@ func snapshotVolumeEngine(ctx context.Context, cfg *Config, volumeID, snapshotID
if err != nil {
return nil, fmt.Errorf("new viperblock: %w", err)
}
defer func() {
vb.StopChunkUploader()
vb.StopWALSyncer()
}()
defer vb.Detach()
if err := vb.Backend.InitCtx(ctx); err != nil {
return nil, fmt.Errorf("backend init: %w", err)
}
@@ -1287,8 +1279,7 @@ func constructMountedVB(ctx context.Context, cfg *Config, volumeName string) (*v
opened := false
defer func() {
if !opened {
vb.StopChunkUploader()
vb.StopWALSyncer()
vb.Detach()
}
}()
@@ -1603,8 +1594,7 @@ func unmountVolume(ctx context.Context, cfg *Config, volumeName string) (types.E
// 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()
matched.VB.Detach()
}
if err := utils.KillProcess(matched.PID); err != nil {
spinifex/services/viperblockd/viperblockd.go
+6 -9
View File
@@ -355,8 +355,7 @@ func openVolumeVB(ctx context.Context, cfg *Config, volumeName string) (*viperbl
opened := false
defer func() {
if !opened {
vb.StopChunkUploader()
vb.StopWALSyncer()
vb.Detach()
cfg.releaseVolumeLease(ctx, lease)
}
}()
@@ -494,7 +493,7 @@ func clearStaleSealReceipt(baseDir, volume string) {
// RecoverLocalWALs. Skipping LoadBlockState would leave an empty in-memory block
// map that Close() then flushes over the good checkpoint in predastore — silent
// data loss (a reattach then finds an empty map, bad superblock). The caller
// MUST Close the returned VB; on error the WAL syncer is stopped and no VB is
// MUST Close the returned VB; on error the engine is detached and no VB is
// returned. The caller MUST ensure no nbdkit process is writing the shared
// BaseDir first (post-KillProcess, or volume detached).
func openLoadedVolumeVB(ctx context.Context, cfg *Config, volumeName string) (*viperblock.VB, *volumeLease, error) {
@@ -503,14 +502,14 @@ func openLoadedVolumeVB(ctx context.Context, cfg *Config, volumeName string) (*v
return nil, nil, err
}
if err := vb.LoadBlockStateCtx(ctx); err != nil {
vb.StopWALSyncer()
vb.Detach()
cfg.releaseVolumeLease(ctx, lease)
return nil, nil, fmt.Errorf("load block state: %w", err)
}
// RecoverLocalWALs is fail-closed on integrity errors and persists recovered
// state itself; on failure retain the local WAL (no Close) for retry.
if err := vb.RecoverLocalWALs(); err != nil {
vb.StopWALSyncer()
vb.Detach()
cfg.releaseVolumeLease(ctx, lease)
return nil, nil, fmt.Errorf("recover local WALs: %w", err)
}
@@ -688,8 +687,7 @@ func launchService(cfg *Config) (err error) {
}
// Stop background goroutines and kill nbdkit process
if matched.VB != nil {
matched.VB.StopChunkUploader()
matched.VB.StopWALSyncer()
matched.VB.Detach()
}
if err := utils.KillProcess(matched.PID); err != nil {
slog.ErrorContext(ctx, "Failed to kill nbdkit process", "pid", matched.PID, "err", err)
@@ -973,8 +971,7 @@ func shutdownVolumes(volumes []MountedVolume, inUse func(MountedVolume) bool) {
var wg sync.WaitGroup
for _, volume := range volumes {
if volume.VB != nil {
volume.VB.StopChunkUploader()
volume.VB.StopWALSyncer()
volume.VB.Detach()
}
if inUse(volume) {
slog.Warn("nbdkit still serving a guest; leaving it for the drain/unmount path",