Lock DATA_DIR against a second instance (closes #201)

Nothing stopped two processes opening the same DATA_DIR. Both open the
same per-webhook databases, both run delivery recovery over the same
rows, and both deliver: every pending delivery reaches the destination
twice, from nothing worse than an overlapping deploy.

The entry point now takes an exclusive advisory flock(2) on
{DATA_DIR}/webhooker.lock before anything opens a database, and holds it
for the process lifetime. A second process pointed at the same directory
prints a message naming that directory and exits 1. The lock is the
kernel's, not the file's, so a process killed with SIGKILL leaves a lock
file that blocks nothing -- which is what a pidfile would get wrong. The
file is never unlinked: doing so would let the next process lock a fresh
inode while a third still held the old one.

Acquisition lives in internal/datadir rather than in the server's fx
graph, so any entry point touching DATA_DIR takes it the same way, and
ErrLocked lets a caller tell a live deployment from any other failure.
config.DataDir() resolves DATA_DIR once, for both the lock and Config,
so the two cannot disagree.

Regression coverage: a real second process is refused, and a restart
after kill -9 succeeds with the stale lock file in place.

github.com/gofrs/flock carries the lock; its own module minimums pull
testify to v1.11.1 and golang.org/x/sys to v0.37.0.

internal/config/config.go

@@ -26,6 +26,10 @@ const (
 	// EnvironmentProd represents production environment.
 	EnvironmentProd = "prod"
 
+	// DefaultDataDir is where all SQLite databases live when DATA_DIR
+	// is unset. The same default applies in every environment.
+	DefaultDataDir = "/var/lib/webhooker"
+
 	// defaultPort is the default HTTP listen port.
 	defaultPort = 8080
 
@@ -134,6 +138,19 @@ func envString(key string) string {
 	return os.Getenv(key)
 }
 
+// DataDir resolves DATA_DIR, applying DefaultDataDir when it is unset
+// or empty. It is exported so that entry points which must act on the
+// data directory before the fx graph exists — taking the exclusive
+// directory lock, above all — resolve it exactly as Config does.
+func DataDir() string {
+	dir := envString("DATA_DIR")
+	if dir == "" {
+		return DefaultDataDir
+	}
+
+	return dir
+}
+
 // envBool returns the value of the named environment variable
 // parsed as a boolean. Returns defaultValue if not set. If the
 // variable is set but cannot be parsed, it returns a wrapped error
@@ -407,7 +424,7 @@ func loadFromEnv() (*Config, error) {
 	}
 
 	return &Config{
-		DataDir:                envString("DATA_DIR"),
+		DataDir:                DataDir(),
 		Debug:                  debug,
 		MaintenanceMode:        maintenanceMode,
 		Environment:            environment,
@@ -485,14 +502,6 @@ func New(lc fx.Lifecycle, params ConfigParams) (*Config, error) {
 	s.log = log
 	s.params = &params
 
-	// Set default DataDir. All SQLite databases (main application
-	// DB and per-webhook event DBs) live here. The same default is
-	// used regardless of environment; override with DATA_DIR if
-	// needed.
-	if s.DataDir == "" {
-		s.DataDir = "/var/lib/webhooker"
-	}
-
 	if s.Debug {
 		params.Logger.EnableDebugLogging()
 	}

internal/config/config_test.go

@@ -418,6 +418,36 @@ func TestDefaultDataDir(t *testing.T) {
 	}
 }
 
+// TestDataDirHelper pins the exported resolver against the value
+// Config carries. The entry point takes the DATA_DIR lock through the
+// helper before the fx graph exists, so the two disagreeing would mean
+// locking one directory and writing to another.
+func TestDataDirHelper(t *testing.T) {
+	for _, set := range []string{"", "/tmp/webhooker-datadir-helper"} {
+		name := "set"
+		if set == "" {
+			name = "unset"
+		}
+
+		t.Run(name, func(t *testing.T) {
+			// Cannot use t.Parallel() here because t.Setenv
+			// is incompatible with parallel subtests.
+			if set == "" {
+				require.NoError(t, os.Unsetenv("DATA_DIR"))
+			} else {
+				t.Setenv("DATA_DIR", set)
+			}
+
+			expected := set
+			if expected == "" {
+				expected = config.DefaultDataDir
+			}
+
+			assert.Equal(t, expected, config.DataDir())
+		})
+	}
+}
+
 func TestReceiverRateLimit(t *testing.T) {
 	tests := []struct {
 		name        string

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 {
+		// TryLock leaves the descriptor open when it fails to take
+		// the lock, so it has to be closed explicitly.
+		_ = 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)
+}