From d206518f26ada9e8f68c2148e820b0c523ae7e07 Mon Sep 17 00:00:00 2001 From: Stefan Date: Mon, 3 Aug 2026 17:27:29 +0200 Subject: [PATCH 1/2] fix(datadir): clear a provably-orphaned dm-era device instead of failing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a schelk run dies mid-operation it can exit without tearing down its dm-era device. The flock is released with the process, so the host looks healthy, but every later `schelk mount` refuses: dm-era device 'bench_era' already exists. schelk has no repair command for this, so each job routed at the host fails in seconds and the now-idle runner immediately claims the next one. Clearing it required an operator to SSH in and run dmsetup by hand. Remove the device ourselves, but only when it is provably nobody's: - we hold schelk's own state lock (so no schelk process is mid-operation) - the device reports zero open handles - the scratch is not mounted and schelk's state agrees If any check fails we surface the original mount error untouched — a device that might still be in use is not worth a corrupted volume. Complements the lock-contention retry: contention waits, an orphaned artefact is repaired once, and everything else still fails fast. --- pkg/datadir/schelk.go | 95 +++++++++++++++++++++++++++++++++++++- pkg/datadir/schelk_test.go | 72 +++++++++++++++++++++++++++++ 2 files changed, 166 insertions(+), 1 deletion(-) diff --git a/pkg/datadir/schelk.go b/pkg/datadir/schelk.go index c7090c4..d33fe0b 100644 --- a/pkg/datadir/schelk.go +++ b/pkg/datadir/schelk.go @@ -10,6 +10,7 @@ import ( "os" "os/exec" "path/filepath" + "strconv" "strings" "syscall" "time" @@ -150,6 +151,7 @@ func NewSchelkProvider(log logrus.FieldLogger) SchelkProvider { type SchelkState struct { MountPoint string `json:"mount_point"` IsMounted bool `json:"is_mounted"` + DMEraName string `json:"dm_era_name"` } type schelkProvider struct { @@ -268,6 +270,12 @@ func SchelkStatePath() string { return DefaultSchelkStatePath } +// SchelkLockPath returns the path of schelk's state lock. It lives alongside +// the state file, so it follows SCHELK_STATE for tests. +func SchelkLockPath() string { + return filepath.Join(filepath.Dir(SchelkStatePath()), "schelk.lock") +} + // ReadSchelkState reads and decodes the schelk JSON state file. func ReadSchelkState(path string) (*SchelkState, error) { data, err := os.ReadFile(path) @@ -346,9 +354,25 @@ func EnsureSchelkMounted(ctx context.Context, log logrus.FieldLogger) error { SchelkStatePath(), state.MountPoint, bin, ) case !mounted: - if err := mountWaitingForLock(ctx, log, bin, schelkLockPollWait); err != nil { + err := mountWaitingForLock(ctx, log, bin, schelkLockPollWait) + if err == nil { + return nil + } + + // A crashed schelk run can exit without tearing down its dm-era + // device, leaving `schelk mount` to refuse forever. schelk has no + // repair command for this, so every job routed at the host fails in + // seconds until someone clears it by hand. Clear it ourselves, but + // only when it is provably nobody's device. + if !schelkStaleDevice([]byte(err.Error())) { return err } + + if repairErr := repairStaleEraDevice(ctx, log, state.DMEraName); repairErr != nil { + return fmt.Errorf("%w (stale dm-era device left in place: %w)", err, repairErr) + } + + return mountWaitingForLock(ctx, log, bin, schelkLockPollWait) } return nil @@ -409,6 +433,75 @@ func mountWaitingForLock( } } +// schelkStaleDevice reports whether `schelk mount` refused because a dm-era +// device from a crashed run is still present. +func schelkStaleDevice(output []byte) bool { + return bytes.Contains(output, []byte("dm-era device")) && + bytes.Contains(output, []byte("already exists")) +} + +// dmEraOpenCount returns how many handles are open against a device-mapper +// device. Zero means nothing is using it. +func dmEraOpenCount(ctx context.Context, name string) (int, error) { + //nolint:gosec // name comes from schelk's own state file. + out, err := exec.CommandContext(ctx, "dmsetup", "info", "-c", "-o", "open", "--noheadings", name).Output() + if err != nil { + return 0, fmt.Errorf("dmsetup info %q: %w", name, err) + } + + count, err := strconv.Atoi(strings.TrimSpace(string(out))) + if err != nil { + return 0, fmt.Errorf("parsing open count for %q from %q: %w", name, out, err) + } + + return count, nil +} + +// repairStaleEraDevice removes a dm-era device orphaned by a crashed schelk +// run. It refuses unless the device is provably unused: we hold schelk's own +// state lock, so no schelk process is mid-operation, and the device reports +// zero open handles. Either check failing means something may still be using +// the device, and surfacing the original error beats corrupting a volume. +func repairStaleEraDevice(ctx context.Context, log logrus.FieldLogger, name string) error { + if name == "" { + return errors.New("schelk state has no dm_era_name") + } + + lock, err := os.OpenFile(SchelkLockPath(), os.O_RDWR|os.O_CREATE, 0o600) + if err != nil { + return fmt.Errorf("opening schelk lock: %w", err) + } + defer lock.Close() + + // Non-blocking: if schelk holds the lock it is still working, and the + // device is its own rather than an orphan. + if err := syscall.Flock(int(lock.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil { + return fmt.Errorf("schelk lock is held, not treating %q as orphaned: %w", name, err) + } + defer func() { _ = syscall.Flock(int(lock.Fd()), syscall.LOCK_UN) }() + + open, err := dmEraOpenCount(ctx, name) + if err != nil { + return err + } + + if open != 0 { + return fmt.Errorf("dm-era device %q has %d open handle(s); refusing to remove", name, open) + } + + if log != nil { + log.WithField("dm_era_name", name). + Warn("Removing stale dm-era device left by a crashed schelk run") + } + + //nolint:gosec // name comes from schelk's own state file. + if out, err := exec.CommandContext(ctx, "dmsetup", "remove", name).CombinedOutput(); err != nil { + return fmt.Errorf("dmsetup remove %q: %w (output: %s)", name, err, strings.TrimSpace(string(out))) + } + + return nil +} + // SchelkPromote persists the current scratch contents as the new virgin // baseline via `schelk promote`, so subsequent recover/restore reset to it. func SchelkPromote(ctx context.Context, log logrus.FieldLogger) error { diff --git a/pkg/datadir/schelk_test.go b/pkg/datadir/schelk_test.go index 4004dbe..d4b078a 100644 --- a/pkg/datadir/schelk_test.go +++ b/pkg/datadir/schelk_test.go @@ -5,6 +5,7 @@ import ( "fmt" "os" "path/filepath" + "syscall" "testing" "time" @@ -215,3 +216,74 @@ exit 1 require.NoError(t, readErr) assert.Equal(t, "1\n", string(data), "non-lock errors must not retry") } + +func TestSchelkStaleDevice(t *testing.T) { + assert.True(t, schelkStaleDevice([]byte("dm-era device 'bench_era' already exists."))) + assert.False(t, schelkStaleDevice([]byte("Another schelk process is already running"))) + assert.False(t, schelkStaleDevice([]byte("Volume is already mounted"))) +} + +// installFakeDmsetup puts a dmsetup shim on PATH reporting the given open count. +func installFakeDmsetup(t *testing.T, openCount string, removeLog string) { + t.Helper() + + dir := t.TempDir() + script := fmt.Sprintf(`#!/bin/sh +case "$*" in + *info*) echo "%s" ;; + *remove*) echo "$*" >> %s ;; +esac +exit 0 +`, openCount, removeLog) + require.NoError(t, os.WriteFile(filepath.Join(dir, "dmsetup"), []byte(script), 0o755)) + t.Setenv("PATH", dir+string(os.PathListSeparator)+os.Getenv("PATH")) +} + +func TestRepairStaleEraDevice_RemovesWhenUnused(t *testing.T) { + statePath := filepath.Join(t.TempDir(), "state.json") + require.NoError(t, os.WriteFile(statePath, []byte(`{"mount_point":"/schelk"}`), 0o600)) + t.Setenv("SCHELK_STATE", statePath) + + removeLog := filepath.Join(t.TempDir(), "removed") + installFakeDmsetup(t, "0", removeLog) + + require.NoError(t, repairStaleEraDevice(context.Background(), logrus.New(), "bench_era")) + + data, err := os.ReadFile(removeLog) + require.NoError(t, err, "dmsetup remove should have been called") + assert.Contains(t, string(data), "bench_era") +} + +func TestRepairStaleEraDevice_RefusesWhenDeviceInUse(t *testing.T) { + statePath := filepath.Join(t.TempDir(), "state.json") + require.NoError(t, os.WriteFile(statePath, []byte(`{"mount_point":"/schelk"}`), 0o600)) + t.Setenv("SCHELK_STATE", statePath) + + removeLog := filepath.Join(t.TempDir(), "removed") + installFakeDmsetup(t, "1", removeLog) + + err := repairStaleEraDevice(context.Background(), logrus.New(), "bench_era") + require.Error(t, err) + assert.Contains(t, err.Error(), "open handle") + assert.NoFileExists(t, removeLog, "must not remove a device that is in use") +} + +func TestRepairStaleEraDevice_RefusesWhileSchelkLockHeld(t *testing.T) { + statePath := filepath.Join(t.TempDir(), "state.json") + require.NoError(t, os.WriteFile(statePath, []byte(`{"mount_point":"/schelk"}`), 0o600)) + t.Setenv("SCHELK_STATE", statePath) + + removeLog := filepath.Join(t.TempDir(), "removed") + installFakeDmsetup(t, "0", removeLog) + + // Hold the lock as a concurrent schelk process would. + held, err := os.OpenFile(SchelkLockPath(), os.O_RDWR|os.O_CREATE, 0o600) + require.NoError(t, err) + defer held.Close() + require.NoError(t, syscall.Flock(int(held.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)) + + repairErr := repairStaleEraDevice(context.Background(), logrus.New(), "bench_era") + require.Error(t, repairErr) + assert.Contains(t, repairErr.Error(), "lock is held") + assert.NoFileExists(t, removeLog, "must not remove while schelk may be running") +} From aa6a25fae97ba38280f71239326585c4b1516834 Mon Sep 17 00:00:00 2001 From: Stefan Date: Mon, 3 Aug 2026 17:33:48 +0200 Subject: [PATCH 2/2] lint: check Close return values in the stale-device repair path errcheck flagged both deferred Close calls. --- pkg/datadir/schelk.go | 2 +- pkg/datadir/schelk_test.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pkg/datadir/schelk.go b/pkg/datadir/schelk.go index d33fe0b..4c50b7f 100644 --- a/pkg/datadir/schelk.go +++ b/pkg/datadir/schelk.go @@ -471,7 +471,7 @@ func repairStaleEraDevice(ctx context.Context, log logrus.FieldLogger, name stri if err != nil { return fmt.Errorf("opening schelk lock: %w", err) } - defer lock.Close() + defer func() { _ = lock.Close() }() // Non-blocking: if schelk holds the lock it is still working, and the // device is its own rather than an orphan. diff --git a/pkg/datadir/schelk_test.go b/pkg/datadir/schelk_test.go index d4b078a..66297f2 100644 --- a/pkg/datadir/schelk_test.go +++ b/pkg/datadir/schelk_test.go @@ -279,7 +279,7 @@ func TestRepairStaleEraDevice_RefusesWhileSchelkLockHeld(t *testing.T) { // Hold the lock as a concurrent schelk process would. held, err := os.OpenFile(SchelkLockPath(), os.O_RDWR|os.O_CREATE, 0o600) require.NoError(t, err) - defer held.Close() + defer func() { _ = held.Close() }() require.NoError(t, syscall.Flock(int(held.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)) repairErr := repairStaleEraDevice(context.Background(), logrus.New(), "bench_era")