Take an exclusive lock on DATA_DIR at startup (closes #201) (#220)

internal/datadir/lock.go

@@ -0,0 +1,126 @@
+// Package datadir guards exclusive access to the directory holding
+// every SQLite database webhooker writes.
+//
+// Two processes sharing a DATA_DIR each open the same per-webhook
+// event databases and each run delivery recovery over the same rows,
+// so every pending delivery goes out twice. SQLite's own locking does
+// not prevent that: both writers are serialised correctly and both
+// deliver. The only thing that prevents it is refusing to be the
+// second process.
+//
+// The lock lives here rather than in the server's fx graph so that any
+// entry point which touches DATA_DIR — the server, or a CLI
+// subcommand that must not operate on a live deployment's data — takes
+// it the same way.
+package datadir
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+
+	"github.com/gofrs/flock"
+)
+
+// LockFileName is the advisory lock file created inside DATA_DIR. Its
+// contents are never read: the lock is the flock(2) held on the open
+// descriptor, not the file's existence, so a leftover file from a
+// process that was killed with SIGKILL blocks nothing.
+const LockFileName = "webhooker.lock"
+
+// dirPerm is the mode Acquire creates DATA_DIR with. It matches what
+// internal/database uses, since whichever runs first creates it.
+const dirPerm = 0o750
+
+// ErrLocked reports that another live process holds the data
+// directory. Callers that need to know whether a deployment is running
+// — rather than merely failing to start — test for this with
+// errors.Is.
+var ErrLocked = errors.New(
+	"data directory is already in use by another instance",
+)
+
+// ErrNoDir reports that Acquire was given an empty directory.
+var ErrNoDir = errors.New("no data directory given")
+
+// Lock is a held exclusive advisory lock on a data directory. It is
+// valid only while the process that took it lives: the kernel drops it
+// when the descriptor closes, whether that is Release, a normal exit,
+// or a SIGKILL.
+type Lock struct {
+	dir  string
+	file *flock.Flock
+}
+
+// Acquire takes the exclusive advisory lock on dir, creating dir if it
+// does not exist. It never waits: if another process holds the lock it
+// returns an error wrapping ErrLocked and naming dir.
+//
+// The returned Lock must be held for as long as the caller intends to
+// use dir.
+func Acquire(dir string) (*Lock, error) {
+	if dir == "" {
+		return nil, ErrNoDir
+	}
+
+	err := os.MkdirAll(dir, dirPerm)
+	if err != nil {
+		return nil, fmt.Errorf(
+			"creating data directory %s: %w", dir, err,
+		)
+	}
+
+	path := filepath.Join(dir, LockFileName)
+	fl := flock.New(path)
+
+	held, err := fl.TryLock()
+	if err != nil {
+		return nil, fmt.Errorf(
+			"locking data directory %s: %w", dir, err,
+		)
+	}
+
+	if !held {
+		// A no-op on flock v0.13.0, which closes its own descriptor on
+		// a failed TryLock; kept so no version can leak one.
+		_ = fl.Close()
+
+		return nil, fmt.Errorf(
+			"%w: %s (%s). Only one webhooker may use a data "+
+				"directory: two both run delivery recovery over the "+
+				"same rows and both deliver",
+			ErrLocked, dir, path,
+		)
+	}
+
+	return &Lock{dir: dir, file: fl}, nil
+}
+
+// Dir returns the locked directory.
+func (l *Lock) Dir() string {
+	return l.dir
+}
+
+// Path returns the lock file backing the lock.
+func (l *Lock) Path() string {
+	return l.file.Path()
+}
+
+// Release drops the lock and closes the descriptor. It is safe to call
+// more than once.
+//
+// The lock file is deliberately left on disk. Unlinking it would let
+// the next process create and lock a fresh inode while a third still
+// holds the old one, which is the one outcome this package exists to
+// prevent.
+func (l *Lock) Release() error {
+	err := l.file.Unlock()
+	if err != nil {
+		return fmt.Errorf(
+			"releasing lock on data directory %s: %w", l.dir, err,
+		)
+	}
+
+	return nil
+}

internal/datadir/lock_test.go

@@ -0,0 +1,250 @@
+package datadir_test
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"sneak.berlin/go/webhooker/internal/datadir"
+)
+
+// holderEnv names the directory the re-executed test binary should
+// lock and hold. When it is unset the child test does nothing, so an
+// ordinary run is unaffected.
+const holderEnv = "WEBHOOKER_DATADIR_LOCK_HOLDER"
+
+// holderReadyPrefix labels the child's one-line report that it holds
+// the lock, so the parent can find it among the testing package's own
+// output on the same descriptor.
+const holderReadyPrefix = "DATADIR-LOCK-HELD "
+
+// holderReadyTimeout bounds the wait for the child to take the lock.
+// It only has to cover process start on a loaded shared host.
+const holderReadyTimeout = 60 * time.Second
+
+// holderHold is how long the child keeps the lock if nothing kills it.
+// A sleep rather than a bare block, so the runtime's deadlock detector
+// has a pending timer and the child cannot outlive a killed test run
+// by more than this.
+const holderHold = 10 * time.Minute
+
+// TestLockHolder is the child half of the two-process tests below. It
+// takes the lock on the directory named by holderEnv, reports the lock
+// file on standard output, and then holds it until it is killed.
+func TestLockHolder(t *testing.T) {
+	t.Parallel()
+
+	dir := os.Getenv(holderEnv)
+	if dir == "" {
+		return
+	}
+
+	lock, err := datadir.Acquire(dir)
+	require.NoError(t, err)
+
+	// Written to the descriptor directly: the parent reads fd 1, not
+	// the testing package's buffered report.
+	_, err = fmt.Fprintf(
+		os.Stdout, "%s%s\n", holderReadyPrefix, lock.Path(),
+	)
+	require.NoError(t, err)
+
+	time.Sleep(holderHold)
+}
+
+// startHolder re-executes this test binary as a separate process that
+// takes and holds the lock on dir, and returns once that process
+// actually holds it. The child is killed when the test ends.
+func startHolder(t *testing.T, dir string) *exec.Cmd {
+	t.Helper()
+
+	//nolint:gosec // Re-executing this test binary, with a fixed arg.
+	cmd := exec.CommandContext(
+		t.Context(), os.Args[0], "-test.run", "^TestLockHolder$",
+	)
+
+	cmd.Env = append(os.Environ(), holderEnv+"="+dir)
+	cmd.Stderr = os.Stderr
+
+	stdout, err := cmd.StdoutPipe()
+	require.NoError(t, err)
+	require.NoError(t, cmd.Start())
+
+	t.Cleanup(func() {
+		_ = cmd.Process.Kill()
+		_ = cmd.Wait()
+	})
+
+	ready := make(chan string, 1)
+
+	go func() {
+		scanner := bufio.NewScanner(stdout)
+		for scanner.Scan() {
+			after, found := strings.CutPrefix(
+				scanner.Text(), holderReadyPrefix,
+			)
+			if found {
+				ready <- after
+
+				break
+			}
+		}
+
+		close(ready)
+
+		// Keep draining so the child never blocks on a full pipe.
+		_, _ = io.Copy(io.Discard, stdout)
+	}()
+
+	select {
+	case path, ok := <-ready:
+		require.True(
+			t, ok, "holder exited without taking the lock",
+		)
+		require.Equal(t, filepath.Join(dir, datadir.LockFileName), path)
+	case <-time.After(holderReadyTimeout):
+		t.Fatal("timed out waiting for the holder to take the lock")
+	}
+
+	return cmd
+}
+
+// TestSecondInstanceRefused is the regression test for the duplicate
+// delivery this package exists to prevent: a real second process
+// pointed at a data directory a live process already holds must be
+// refused, with an error that names the directory.
+func TestSecondInstanceRefused(t *testing.T) {
+	t.Parallel()
+
+	dir := t.TempDir()
+
+	startHolder(t, dir)
+
+	lock, err := datadir.Acquire(dir)
+
+	require.Error(t, err, "the second instance took the lock too")
+	require.Nil(t, lock)
+	require.ErrorIs(
+		t, err, datadir.ErrLocked,
+		"the refusal must be distinguishable from any other failure",
+	)
+	assert.Contains(
+		t, err.Error(), dir,
+		"the refusal must name the directory it is about",
+	)
+}
+
+// TestRestartAfterHardKill is the other half of the regression: a
+// process killed with SIGKILL runs no cleanup and leaves its lock file
+// behind, and the next start must not be blocked by it. This is what a
+// pidfile would get wrong; the kernel drops a flock when the
+// descriptor closes, however the process died.
+func TestRestartAfterHardKill(t *testing.T) {
+	t.Parallel()
+
+	dir := t.TempDir()
+
+	holder := startHolder(t, dir)
+
+	require.NoError(t, holder.Process.Kill())
+
+	// Wait for the kill to have actually happened. Re-acquiring while
+	// the corpse still holds a descriptor would be a race, and would
+	// make this test pass or fail on scheduling.
+	_ = holder.Wait()
+
+	require.FileExists(
+		t, filepath.Join(dir, datadir.LockFileName),
+		"the stale lock file is what must not block the restart",
+	)
+
+	lock, err := datadir.Acquire(dir)
+	require.NoError(
+		t, err, "a hard-killed instance must not block the next start",
+	)
+
+	require.NoError(t, lock.Release())
+}
+
+// TestSecondFdInSameProcessRefused pins the flock(2) property the
+// tests in cmd/webhooker rely on: descriptors are locked
+// independently, so a second acquisition is denied even when it comes
+// from the process that already holds the lock.
+func TestSecondFdInSameProcessRefused(t *testing.T) {
+	t.Parallel()
+
+	dir := t.TempDir()
+
+	first, err := datadir.Acquire(dir)
+	require.NoError(t, err)
+
+	defer func() { _ = first.Release() }()
+
+	_, err = datadir.Acquire(dir)
+	require.ErrorIs(t, err, datadir.ErrLocked)
+}
+
+// TestReleaseAllowsReacquire covers the clean-shutdown path: the lock
+// is released on exit, so a restart is not blocked by the previous
+// run.
+func TestReleaseAllowsReacquire(t *testing.T) {
+	t.Parallel()
+
+	dir := t.TempDir()
+
+	first, err := datadir.Acquire(dir)
+	require.NoError(t, err)
+	require.NoError(t, first.Release())
+
+	second, err := datadir.Acquire(dir)
+	require.NoError(t, err)
+	require.NoError(t, second.Release())
+}
+
+// TestAcquireCreatesDataDir covers a first start against a DATA_DIR
+// that does not exist yet, which is the normal case for a fresh
+// deployment: the lock is taken before anything else creates it.
+func TestAcquireCreatesDataDir(t *testing.T) {
+	t.Parallel()
+
+	dir := filepath.Join(t.TempDir(), "nested", "data")
+
+	lock, err := datadir.Acquire(dir)
+	require.NoError(t, err)
+
+	defer func() { _ = lock.Release() }()
+
+	assert.Equal(t, dir, lock.Dir())
+	assert.FileExists(t, filepath.Join(dir, datadir.LockFileName))
+}
+
+// TestAcquireEmptyDir rejects an empty directory rather than locking
+// the process's working directory.
+func TestAcquireEmptyDir(t *testing.T) {
+	t.Parallel()
+
+	_, err := datadir.Acquire("")
+	require.ErrorIs(t, err, datadir.ErrNoDir)
+}
+
+// TestAcquireUnusableDir reports an unusable DATA_DIR clearly, naming
+// it, instead of failing later and deeper.
+func TestAcquireUnusableDir(t *testing.T) {
+	t.Parallel()
+
+	file := filepath.Join(t.TempDir(), "not-a-directory")
+	require.NoError(t, os.WriteFile(file, nil, 0o600))
+
+	_, err := datadir.Acquire(file)
+
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), file)
+}