diff --git a/pkg/datadir/schelk.go b/pkg/datadir/schelk.go index c7090c4..4c50b7f 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 func() { _ = 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..66297f2 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 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") + require.Error(t, repairErr) + assert.Contains(t, repairErr.Error(), "lock is held") + assert.NoFileExists(t, removeLog, "must not remove while schelk may be running") +}