From f4dc1dbe54e2cd84845ca9f5446b638c79453317 Mon Sep 17 00:00:00 2001
From: theluckystrike <51033404+theluckystrike@users.noreply.github.com>
Date: Wed, 7 Oct 2026 02:20:26 +0000
Subject: [PATCH 1/2] Reuse a stale non-completed snapshot request when
 snapshots are repeatable

A run killed mid-snapshot leaves a "requested" or "in progress" row in
pgstream.snapshot_requests. The partial unique index
schema_table_status_hash_unique_index allows one non-completed row per
schema and table list. So every later run fails at once in
CreateSnapshotRequest with a constraint violation. With repeatable
snapshots on, the recorder never reads the existing rows, so the only
way out was to delete the row by hand (xataio/pgstream#1222).

When CreateSnapshotRequest returns a constraint violation:
- With repeatable snapshots on, log a warning and reuse the row. The
  existing MarkInProgress and UpdateSnapshotRequest calls already match
  the non-completed row by schema, tables and mode.
- With repeatable snapshots off, return an error that names the table
  and tells the user what to do. The original error stays wrapped.

Other errors are returned as before. No migration: the mode column
missing from the unique index (issue point 3) is out of scope.

Tests (go1.26.6 darwin/arm64)

go test -count=1 -run 'TestSnapshotRecorder_CreateSnapshot_(repeatableReusesStaleRequest|nonRepeatableConflictIsActionable)' -v ./pkg/snapshot/generator/

Before the fix (main code, new tests only):
  Error: Received unexpected error:
         error creating snapshot request: constraint violation: duplicate key value violates unique constraint "schema_table_status_hash_unique_index"
  --- FAIL: TestSnapshotRecorder_CreateSnapshot_repeatableReusesStaleRequest (0.00s)
  Error: Error "error creating snapshot request: constraint violation: duplicate key value violates unique constraint \"schema_table_status_hash_unique_index\"" does not contain "pgstream.snapshot_requests"
  --- FAIL: TestSnapshotRecorder_CreateSnapshot_nonRepeatableConflictIsActionable (0.00s)
  FAIL	github.com/xataio/pgstream/pkg/snapshot/generator	0.507s

After the fix:
  --- PASS: TestSnapshotRecorder_CreateSnapshot_nonRepeatableConflictIsActionable (0.00s)
  --- PASS: TestSnapshotRecorder_CreateSnapshot_repeatableReusesStaleRequest (0.00s)
  ok  	github.com/xataio/pgstream/pkg/snapshot/generator	1.683s

End to end, the issue's steps against a local PostgreSQL 16.13:
`pgstream snapshot` with repeatable_snapshots: true, mode data,
public.t1 with 3,000,000 rows, recorder store and target in the same
database. Run 1 is killed with kill -9 once its request row is
"in progress" and the target has rows. Run 2 is the same snapshot into
the truncated target.
  binary built from main:
    run 1 SIGKILL, row 3 public {t1} "in progress", target 3000 rows
    run 2 exit 1
    Error: error creating snapshot request: constraint violation: duplicate key value violates unique constraint "schema_table_status_hash_unique_index"
    row stays "in progress", target 0 rows
  binary built with this fix:
    run 1 SIGKILL, row 5 public {t1} "in progress", target 21000 rows
    run 2 exit 0
    WRN reusing non-completed snapshot request left by a previous run mode=data schema=public
    row 5 ends "completed" with no errors, target 3000000 rows
A leftover "requested" row behaves the same (main exit 1, fix exit 0 and
the row ends "completed"), and a repeatable run after that one adds a
new completed row as before.

Other gates run on this commit:
  go test -count=1 -race ./pkg/snapshot/...                  ok
  go vet ./pkg/snapshot/...                                  ok
  TZ=UTC go test -timeout 10m -race -cover ./... (make test) ok, 56 packages
  make generate                                              no diff
  golangci-lint v2.10.1 run --config=.golangci.yml           0 issues
  license-header-checker -a -r ./license-header.txt . go     495 licenses ok

Not run: make integration-test (testcontainers needs Docker, which is
not available here).
---
 .../generator/snapshot_generator_recorder.go  | 18 ++++-
 .../snapshot_generator_recorder_test.go       | 81 +++++++++++++++++++
 2 files changed, 98 insertions(+), 1 deletion(-)

diff --git a/pkg/snapshot/generator/snapshot_generator_recorder.go b/pkg/snapshot/generator/snapshot_generator_recorder.go
index 0b6d8578d..6fafd082c 100644
--- a/pkg/snapshot/generator/snapshot_generator_recorder.go
+++ b/pkg/snapshot/generator/snapshot_generator_recorder.go
@@ -9,6 +9,7 @@ import (
 	"slices"
 	"time"
 
+	pglib "github.com/xataio/pgstream/internal/postgres"
 	"github.com/xataio/pgstream/pkg/backoff"
 	loglib "github.com/xataio/pgstream/pkg/log"
 	"github.com/xataio/pgstream/pkg/snapshot"
@@ -160,7 +161,22 @@ func (s *SnapshotRecorder) markSnapshotInProgress(ctx context.Context, requests
 	for _, req := range requests {
 		eg.Go(func() error {
 			if err := s.store.CreateSnapshotRequest(ctx, req); err != nil {
-				return err
+				conflictErr := &pglib.ErrConstraintViolation{}
+				if !errors.As(err, &conflictErr) {
+					return err
+				}
+				// a non-completed request for the same schema and tables
+				// already exists, usually left behind by a run that was
+				// killed mid-snapshot
+				if !s.repeatableSnapshots {
+					return fmt.Errorf("a non-completed snapshot request for schema %q already exists in %s.%s; if no snapshot is running, delete that row or mark it completed: %w", req.Schema, snapshotstore.SchemaName, snapshotstore.TableName, err)
+				}
+				// repeatable snapshots run again regardless of existing
+				// requests, so reuse the stale one instead of failing
+				s.logger.Warn(err, "reusing non-completed snapshot request left by a previous run", loglib.Fields{
+					"schema": req.Schema,
+					"mode":   req.GetMode(),
+				})
 			}
 			// the snapshot will start immediately
 			req.MarkInProgress()
diff --git a/pkg/snapshot/generator/snapshot_generator_recorder_test.go b/pkg/snapshot/generator/snapshot_generator_recorder_test.go
index 2cf5339ba..bc1b4fd4a 100644
--- a/pkg/snapshot/generator/snapshot_generator_recorder_test.go
+++ b/pkg/snapshot/generator/snapshot_generator_recorder_test.go
@@ -10,6 +10,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/require"
+	pglib "github.com/xataio/pgstream/internal/postgres"
 	"github.com/xataio/pgstream/pkg/snapshot"
 	snapshotstore "github.com/xataio/pgstream/pkg/snapshot/store"
 	snapshotstoremocks "github.com/xataio/pgstream/pkg/snapshot/store/mocks"
@@ -1007,3 +1008,83 @@ func TestSnapshotRecorder_filterOutExistingSnapshots(t *testing.T) {
 		})
 	}
 }
+
+func TestSnapshotRecorder_CreateSnapshot_repeatableReusesStaleRequest(t *testing.T) {
+	t.Parallel()
+
+	// a previous run was killed mid-snapshot and left a non-completed request
+	// behind, so creating a new one hits the partial unique index
+	dupErr := &pglib.ErrConstraintViolation{
+		Details: `duplicate key value violates unique constraint "schema_table_status_hash_unique_index"`,
+	}
+
+	var statuses []snapshot.Status
+	store := &snapshotstoremocks.Store{
+		CreateSnapshotRequestFn: func(ctx context.Context, r *snapshot.Request) error {
+			return fmt.Errorf("error creating snapshot request: %w", dupErr)
+		},
+		UpdateSnapshotRequestFn: func(ctx context.Context, _ uint, r *snapshot.Request) error {
+			statuses = append(statuses, r.Status)
+			return nil
+		},
+	}
+
+	generatorCalled := false
+	generator := &mockGenerator{
+		createSnapshotFn: func(ctx context.Context, ss *snapshot.Snapshot) error {
+			generatorCalled = true
+			return nil
+		},
+	}
+
+	sr := NewSnapshotRecorder(&Config{
+		RepeatableSnapshots: true,
+		SnapshotWorkers:     1,
+	}, store, generator)
+	defer sr.Close()
+
+	err := sr.CreateSnapshot(context.Background(), &snapshot.Snapshot{
+		SchemaTables: map[string][]string{"public": {"t1"}},
+	})
+	require.NoError(t, err)
+	require.True(t, generatorCalled)
+	require.Equal(t, []snapshot.Status{snapshot.StatusInProgress, snapshot.StatusCompleted}, statuses)
+}
+
+func TestSnapshotRecorder_CreateSnapshot_nonRepeatableConflictIsActionable(t *testing.T) {
+	t.Parallel()
+
+	dupErr := &pglib.ErrConstraintViolation{
+		Details: `duplicate key value violates unique constraint "schema_table_status_hash_unique_index"`,
+	}
+
+	store := &snapshotstoremocks.Store{
+		GetSnapshotRequestsBySchemaFn: func(ctx context.Context, s string) ([]*snapshot.Request, error) {
+			return []*snapshot.Request{}, nil
+		},
+		CreateSnapshotRequestFn: func(ctx context.Context, r *snapshot.Request) error {
+			return fmt.Errorf("error creating snapshot request: %w", dupErr)
+		},
+		UpdateSnapshotRequestFn: func(ctx context.Context, _ uint, r *snapshot.Request) error {
+			return fmt.Errorf("unexpected call to UpdateSnapshotRequestFn: %v", r)
+		},
+	}
+
+	generator := &mockGenerator{
+		createSnapshotFn: func(ctx context.Context, ss *snapshot.Snapshot) error {
+			return errors.New("unexpected call to CreateSnapshot")
+		},
+	}
+
+	sr := NewSnapshotRecorder(&Config{
+		RepeatableSnapshots: false,
+		SnapshotWorkers:     1,
+	}, store, generator)
+	defer sr.Close()
+
+	err := sr.CreateSnapshot(context.Background(), &snapshot.Snapshot{
+		SchemaTables: map[string][]string{"public": {"t1"}},
+	})
+	require.ErrorIs(t, err, dupErr)
+	require.ErrorContains(t, err, "pgstream.snapshot_requests")
+}

From 1fdc11bb65a71df18f4314dfb15a8138f09b1203 Mon Sep 17 00:00:00 2001
From: theluckystrike <51033404+theluckystrike@users.noreply.github.com>
Date: Mon, 19 Oct 2026 07:17:12 +0700
Subject: [PATCH 2/2] Only reuse a stale snapshot request when its run has
 ended

Review on #1237 pointed out that a duplicate row does not prove the old
run stopped. Checked against real PostgreSQL 16 on f4dc1db:
- a retry started while a slow run was still copying went ahead, and
  its completion marked the shared row completed before the slow run
  finished
- a leftover schema-only row for the same tables made a data run
  report success while UpdateSnapshotRequest (keyed on mode) matched
  nothing, so the row stayed "requested"
- every class 23 error was taken for the duplicate

Changes:
- Store.LockSnapshotRequests takes a session level advisory lock per
  request on a dedicated connection, keyed like the unique index
  (schema name and table names hash, no mode). The recorder holds it
  for the whole run. A second run on the same scope fails with
  ErrSnapshotRequestLocked and touches nothing. The server frees the
  lock when the holder's connection ends, so a killed run's request
  can be recovered.
- Store.ResumeSnapshotRequest takes the leftover row over by schema
  and table names whatever its mode, sets it in progress with the new
  mode and reports whether a row matched. When none matched (the row
  completed between the insert and the take over) the recorder
  inserts again.
- CreateSnapshotRequest wraps ErrNonCompletedRequestExists only for a
  unique violation on schema_table_status_hash_unique_index. Other
  constraint errors are returned as before.

Tests:
- unit tests for the recorder (take over, insert again after the gap,
  take over error, non repeatable conflict, other constraint error,
  lock held, lock released on failure) and for the store methods
- integration tests against real postgres: lock exclusivity across
  modes, lock freed when the holder's backend is terminated, take over
  across modes and completion update, and two retries started together
  while a slow run copies (both refused, no copy, the slow run's row
  stays in progress until it completes), then recovery of a dead run
---
 .../generator/snapshot_generator_recorder.go  |  78 ++++--
 .../snapshot_generator_recorder_test.go       | 212 ++++++++++-----
 .../instrumented_snapshot_store.go            |  15 ++
 .../store/mocks/mock_snapshot_store.go        |  16 +-
 ...napshot_store_recovery_integration_test.go | 255 ++++++++++++++++++
 .../store/postgres/pg_snapshot_store.go       |  83 +++++-
 .../store/postgres/pg_snapshot_store_test.go  | 158 +++++++++++
 pkg/snapshot/store/snapshot_store.go          |  19 ++
 8 files changed, 745 insertions(+), 91 deletions(-)
 create mode 100644 pkg/snapshot/store/postgres/integration/pg_snapshot_store_recovery_integration_test.go

diff --git a/pkg/snapshot/generator/snapshot_generator_recorder.go b/pkg/snapshot/generator/snapshot_generator_recorder.go
index 6fafd082c..0b42251e4 100644
--- a/pkg/snapshot/generator/snapshot_generator_recorder.go
+++ b/pkg/snapshot/generator/snapshot_generator_recorder.go
@@ -9,7 +9,6 @@ import (
 	"slices"
 	"time"
 
-	pglib "github.com/xataio/pgstream/internal/postgres"
 	"github.com/xataio/pgstream/pkg/backoff"
 	loglib "github.com/xataio/pgstream/pkg/log"
 	"github.com/xataio/pgstream/pkg/snapshot"
@@ -83,11 +82,24 @@ func (s *SnapshotRecorder) CreateSnapshot(ctx context.Context, ss *snapshot.Snap
 
 	requests := s.createRequests(ss)
 
+	// only one run at a time may work on a given schema and table list. A
+	// leftover non-completed request is reused below only while holding this
+	// lock: a live run would still hold it, since the lock is released only
+	// when that run ends or its process dies.
+	release, err := s.store.LockSnapshotRequests(ctx, requests)
+	if err != nil {
+		if errors.Is(err, snapshotstore.ErrSnapshotRequestLocked) {
+			return fmt.Errorf("another snapshot of the same schema and tables is running, wait for it to finish: %w", err)
+		}
+		return err
+	}
+	defer release()
+
 	if err := s.markSnapshotInProgress(ctx, requests); err != nil {
 		return err
 	}
 
-	err := s.wrapped.CreateSnapshot(ctx, ss)
+	err = s.wrapped.CreateSnapshot(ctx, ss)
 
 	return s.markSnapshotCompleted(ctx, requests, err)
 }
@@ -160,33 +172,53 @@ func (s *SnapshotRecorder) markSnapshotInProgress(ctx context.Context, requests
 	// create one request per schema
 	for _, req := range requests {
 		eg.Go(func() error {
-			if err := s.store.CreateSnapshotRequest(ctx, req); err != nil {
-				conflictErr := &pglib.ErrConstraintViolation{}
-				if !errors.As(err, &conflictErr) {
-					return err
-				}
-				// a non-completed request for the same schema and tables
-				// already exists, usually left behind by a run that was
-				// killed mid-snapshot
-				if !s.repeatableSnapshots {
-					return fmt.Errorf("a non-completed snapshot request for schema %q already exists in %s.%s; if no snapshot is running, delete that row or mark it completed: %w", req.Schema, snapshotstore.SchemaName, snapshotstore.TableName, err)
-				}
-				// repeatable snapshots run again regardless of existing
-				// requests, so reuse the stale one instead of failing
-				s.logger.Warn(err, "reusing non-completed snapshot request left by a previous run", loglib.Fields{
-					"schema": req.Schema,
-					"mode":   req.GetMode(),
-				})
-			}
-			// the snapshot will start immediately
-			req.MarkInProgress()
-			return s.store.UpdateSnapshotRequest(ctx, req)
+			return s.startRequest(ctx, req)
 		})
 	}
 
 	return eg.Wait()
 }
 
+// startRequest records the request as in progress. A non-completed request
+// for the same schema and table list can only be left behind by a run that
+// has ended, since the caller holds the request lock. Repeatable snapshots
+// take it over, otherwise the conflict is reported.
+func (s *SnapshotRecorder) startRequest(ctx context.Context, req *snapshot.Request) error {
+	err := s.store.CreateSnapshotRequest(ctx, req)
+	if err == nil {
+		// the snapshot will start immediately
+		req.MarkInProgress()
+		return s.store.UpdateSnapshotRequest(ctx, req)
+	}
+	if !errors.Is(err, snapshotstore.ErrNonCompletedRequestExists) {
+		return err
+	}
+	if !s.repeatableSnapshots {
+		return fmt.Errorf("a non-completed snapshot request for schema %q already exists in %s.%s; if no snapshot is running, delete that row or mark it completed: %w", req.Schema, snapshotstore.SchemaName, snapshotstore.TableName, err)
+	}
+
+	resumed, resumeErr := s.store.ResumeSnapshotRequest(ctx, req)
+	if resumeErr != nil {
+		return resumeErr
+	}
+	if resumed {
+		s.logger.Warn(err, "reusing non-completed snapshot request left by a previous run", loglib.Fields{
+			"schema": req.Schema,
+			"mode":   req.GetMode(),
+		})
+		req.MarkInProgress()
+		return nil
+	}
+
+	// the leftover request was completed between the insert and the take
+	// over, so the insert can go through now
+	if err := s.store.CreateSnapshotRequest(ctx, req); err != nil {
+		return err
+	}
+	req.MarkInProgress()
+	return s.store.UpdateSnapshotRequest(ctx, req)
+}
+
 func (s *SnapshotRecorder) markSnapshotCompleted(ctx context.Context, requests []*snapshot.Request, err error) error {
 	// make sure we can update the request status in the store regardless of
 	// context cancelations
diff --git a/pkg/snapshot/generator/snapshot_generator_recorder_test.go b/pkg/snapshot/generator/snapshot_generator_recorder_test.go
index bc1b4fd4a..c79413d29 100644
--- a/pkg/snapshot/generator/snapshot_generator_recorder_test.go
+++ b/pkg/snapshot/generator/snapshot_generator_recorder_test.go
@@ -10,7 +10,6 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/require"
-	pglib "github.com/xataio/pgstream/internal/postgres"
 	"github.com/xataio/pgstream/pkg/snapshot"
 	snapshotstore "github.com/xataio/pgstream/pkg/snapshot/store"
 	snapshotstoremocks "github.com/xataio/pgstream/pkg/snapshot/store/mocks"
@@ -1009,82 +1008,163 @@ func TestSnapshotRecorder_filterOutExistingSnapshots(t *testing.T) {
 	}
 }
 
-func TestSnapshotRecorder_CreateSnapshot_repeatableReusesStaleRequest(t *testing.T) {
+// errStaleRequest is what the store returns when a non-completed request for
+// the same schema and tables is already recorded
+var errStaleRequest = fmt.Errorf("error creating snapshot request: %w: %w", snapshotstore.ErrNonCompletedRequestExists,
+	errors.New(`constraint violation: duplicate key value violates unique constraint "schema_table_status_hash_unique_index"`))
+
+func TestSnapshotRecorder_CreateSnapshot_staleRequest(t *testing.T) {
 	t.Parallel()
 
-	// a previous run was killed mid-snapshot and left a non-completed request
-	// behind, so creating a new one hits the partial unique index
-	dupErr := &pglib.ErrConstraintViolation{
-		Details: `duplicate key value violates unique constraint "schema_table_status_hash_unique_index"`,
-	}
+	errTest := errors.New("oh noes")
 
-	var statuses []snapshot.Status
-	store := &snapshotstoremocks.Store{
-		CreateSnapshotRequestFn: func(ctx context.Context, r *snapshot.Request) error {
-			return fmt.Errorf("error creating snapshot request: %w", dupErr)
+	tests := []struct {
+		name        string
+		repeatable  bool
+		createErrs  []error
+		lockErr     error
+		resumed     bool
+		resumeErr   error
+		generateErr error
+
+		wantErr         error
+		wantErrContains string
+		wantCreates     int
+		wantResumes     int
+		wantGenerated   bool
+		wantStatuses    []snapshot.Status
+	}{
+		{
+			name:          "repeatable takes over the stale request",
+			repeatable:    true,
+			createErrs:    []error{errStaleRequest},
+			resumed:       true,
+			wantCreates:   1,
+			wantResumes:   1,
+			wantGenerated: true,
+			wantStatuses:  []snapshot.Status{snapshot.StatusCompleted},
 		},
-		UpdateSnapshotRequestFn: func(ctx context.Context, _ uint, r *snapshot.Request) error {
-			statuses = append(statuses, r.Status)
-			return nil
+		{
+			name:          "repeatable inserts again when the stale request completed before the take over",
+			repeatable:    true,
+			createErrs:    []error{errStaleRequest, nil},
+			resumed:       false,
+			wantCreates:   2,
+			wantResumes:   1,
+			wantGenerated: true,
+			wantStatuses:  []snapshot.Status{snapshot.StatusInProgress, snapshot.StatusCompleted},
 		},
-	}
-
-	generatorCalled := false
-	generator := &mockGenerator{
-		createSnapshotFn: func(ctx context.Context, ss *snapshot.Snapshot) error {
-			generatorCalled = true
-			return nil
+		{
+			name:        "repeatable fails when the take over fails",
+			repeatable:  true,
+			createErrs:  []error{errStaleRequest},
+			resumeErr:   errTest,
+			wantErr:     errTest,
+			wantCreates: 1,
+			wantResumes: 1,
 		},
-	}
-
-	sr := NewSnapshotRecorder(&Config{
-		RepeatableSnapshots: true,
-		SnapshotWorkers:     1,
-	}, store, generator)
-	defer sr.Close()
-
-	err := sr.CreateSnapshot(context.Background(), &snapshot.Snapshot{
-		SchemaTables: map[string][]string{"public": {"t1"}},
-	})
-	require.NoError(t, err)
-	require.True(t, generatorCalled)
-	require.Equal(t, []snapshot.Status{snapshot.StatusInProgress, snapshot.StatusCompleted}, statuses)
-}
-
-func TestSnapshotRecorder_CreateSnapshot_nonRepeatableConflictIsActionable(t *testing.T) {
-	t.Parallel()
-
-	dupErr := &pglib.ErrConstraintViolation{
-		Details: `duplicate key value violates unique constraint "schema_table_status_hash_unique_index"`,
-	}
-
-	store := &snapshotstoremocks.Store{
-		GetSnapshotRequestsBySchemaFn: func(ctx context.Context, s string) ([]*snapshot.Request, error) {
-			return []*snapshot.Request{}, nil
+		{
+			name:            "non repeatable conflict is actionable",
+			repeatable:      false,
+			createErrs:      []error{errStaleRequest},
+			wantErr:         snapshotstore.ErrNonCompletedRequestExists,
+			wantErrContains: "pgstream.snapshot_requests",
+			wantCreates:     1,
 		},
-		CreateSnapshotRequestFn: func(ctx context.Context, r *snapshot.Request) error {
-			return fmt.Errorf("error creating snapshot request: %w", dupErr)
+		{
+			name:        "other constraint errors are not taken for a stale request",
+			repeatable:  true,
+			createErrs:  []error{errTest},
+			wantErr:     errTest,
+			wantCreates: 1,
 		},
-		UpdateSnapshotRequestFn: func(ctx context.Context, _ uint, r *snapshot.Request) error {
-			return fmt.Errorf("unexpected call to UpdateSnapshotRequestFn: %v", r)
+		{
+			name:            "another run holds the lock",
+			repeatable:      true,
+			lockErr:         fmt.Errorf("schema \"public\": %w", snapshotstore.ErrSnapshotRequestLocked),
+			wantErr:         snapshotstore.ErrSnapshotRequestLocked,
+			wantErrContains: "another snapshot of the same schema and tables is running",
 		},
-	}
-
-	generator := &mockGenerator{
-		createSnapshotFn: func(ctx context.Context, ss *snapshot.Snapshot) error {
-			return errors.New("unexpected call to CreateSnapshot")
+		{
+			name:          "lock is released when the snapshot fails",
+			repeatable:    true,
+			createErrs:    []error{nil},
+			generateErr:   errTest,
+			wantErr:       errTest,
+			wantCreates:   1,
+			wantGenerated: true,
+			wantStatuses:  []snapshot.Status{snapshot.StatusInProgress, snapshot.StatusCompleted},
 		},
 	}
 
-	sr := NewSnapshotRecorder(&Config{
-		RepeatableSnapshots: false,
-		SnapshotWorkers:     1,
-	}, store, generator)
-	defer sr.Close()
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
 
-	err := sr.CreateSnapshot(context.Background(), &snapshot.Snapshot{
-		SchemaTables: map[string][]string{"public": {"t1"}},
-	})
-	require.ErrorIs(t, err, dupErr)
-	require.ErrorContains(t, err, "pgstream.snapshot_requests")
+			creates, resumes, locks, releases := 0, 0, 0, 0
+			var statuses []snapshot.Status
+			store := &snapshotstoremocks.Store{
+				LockSnapshotRequestsFn: func(ctx context.Context, reqs []*snapshot.Request) (func(), error) {
+					locks++
+					require.Len(t, reqs, 1)
+					if tc.lockErr != nil {
+						return nil, tc.lockErr
+					}
+					return func() { releases++ }, nil
+				},
+				CreateSnapshotRequestFn: func(ctx context.Context, r *snapshot.Request) error {
+					creates++
+					require.LessOrEqual(t, creates, len(tc.createErrs))
+					return tc.createErrs[creates-1]
+				},
+				ResumeSnapshotRequestFn: func(ctx context.Context, r *snapshot.Request) (bool, error) {
+					resumes++
+					require.Equal(t, "public", r.Schema)
+					require.Equal(t, snapshot.RequestModeData, r.GetMode())
+					return tc.resumed, tc.resumeErr
+				},
+				UpdateSnapshotRequestFn: func(ctx context.Context, _ uint, r *snapshot.Request) error {
+					statuses = append(statuses, r.Status)
+					return nil
+				},
+				GetSnapshotRequestsBySchemaFn: func(ctx context.Context, s string) ([]*snapshot.Request, error) {
+					return []*snapshot.Request{}, nil
+				},
+			}
+
+			generated := false
+			generator := &mockGenerator{
+				createSnapshotFn: func(ctx context.Context, ss *snapshot.Snapshot) error {
+					generated = true
+					return tc.generateErr
+				},
+			}
+
+			sr := NewSnapshotRecorder(&Config{
+				RepeatableSnapshots: tc.repeatable,
+				SnapshotWorkers:     1,
+			}, store, generator)
+			defer sr.Close()
+
+			err := sr.CreateSnapshot(context.Background(), &snapshot.Snapshot{
+				SchemaTables: map[string][]string{"public": {"t1"}},
+			})
+			if tc.wantErr != nil {
+				require.ErrorIs(t, err, tc.wantErr)
+			} else {
+				require.NoError(t, err)
+			}
+			if tc.wantErrContains != "" {
+				require.ErrorContains(t, err, tc.wantErrContains)
+			}
+			require.Equal(t, 1, locks)
+			if tc.lockErr == nil {
+				require.Equal(t, 1, releases)
+			}
+			require.Equal(t, tc.wantCreates, creates)
+			require.Equal(t, tc.wantResumes, resumes)
+			require.Equal(t, tc.wantGenerated, generated)
+			require.Equal(t, tc.wantStatuses, statuses)
+		})
+	}
 }
diff --git a/pkg/snapshot/store/instrumentation/instrumented_snapshot_store.go b/pkg/snapshot/store/instrumentation/instrumented_snapshot_store.go
index 90eac3e2d..afaac64ab 100644
--- a/pkg/snapshot/store/instrumentation/instrumented_snapshot_store.go
+++ b/pkg/snapshot/store/instrumentation/instrumented_snapshot_store.go
@@ -43,6 +43,21 @@ func (i *Store) UpdateSnapshotRequest(ctx context.Context, req *snapshot.Request
 	return i.inner.UpdateSnapshotRequest(ctx, req)
 }
 
+func (i *Store) LockSnapshotRequests(ctx context.Context, reqs []*snapshot.Request) (release func(), err error) {
+	ctx, span := otel.StartSpan(ctx, i.tracer, "snapshotStore.LockSnapshotRequests",
+		trace.WithAttributes(attribute.KeyValue{Key: "requests", Value: attribute.IntValue(len(reqs))}))
+	defer otel.CloseSpan(span, err)
+
+	return i.inner.LockSnapshotRequests(ctx, reqs)
+}
+
+func (i *Store) ResumeSnapshotRequest(ctx context.Context, req *snapshot.Request) (resumed bool, err error) {
+	ctx, span := otel.StartSpan(ctx, i.tracer, "snapshotStore.ResumeSnapshotRequest", trace.WithAttributes(i.requestAttributes(req)...))
+	defer otel.CloseSpan(span, err)
+
+	return i.inner.ResumeSnapshotRequest(ctx, req)
+}
+
 func (i *Store) GetSnapshotRequestsByStatus(ctx context.Context, status snapshot.Status) (req []*snapshot.Request, err error) {
 	ctx, span := otel.StartSpan(ctx, i.tracer, "snapshotStore.GetSnapshotRequestsByStatus",
 		trace.WithAttributes(attribute.KeyValue{Key: "status", Value: attribute.StringValue(string(status))}))
diff --git a/pkg/snapshot/store/mocks/mock_snapshot_store.go b/pkg/snapshot/store/mocks/mock_snapshot_store.go
index d2b8a8e46..b1ca0ba2a 100644
--- a/pkg/snapshot/store/mocks/mock_snapshot_store.go
+++ b/pkg/snapshot/store/mocks/mock_snapshot_store.go
@@ -13,7 +13,10 @@ type Store struct {
 	UpdateSnapshotRequestFn       func(context.Context, uint, *snapshot.Request) error
 	GetSnapshotRequestsByStatusFn func(ctx context.Context, status snapshot.Status) ([]*snapshot.Request, error)
 	GetSnapshotRequestsBySchemaFn func(ctx context.Context, s string) ([]*snapshot.Request, error)
-	updateSnapshotRequestCalls    uint
+	// LockSnapshotRequestsFn defaults to granting every lock when nil
+	LockSnapshotRequestsFn     func(context.Context, []*snapshot.Request) (func(), error)
+	ResumeSnapshotRequestFn    func(context.Context, *snapshot.Request) (bool, error)
+	updateSnapshotRequestCalls uint
 }
 
 func (m *Store) CreateSnapshotRequest(ctx context.Context, s *snapshot.Request) error {
@@ -25,6 +28,17 @@ func (m *Store) UpdateSnapshotRequest(ctx context.Context, s *snapshot.Request)
 	return m.UpdateSnapshotRequestFn(ctx, m.updateSnapshotRequestCalls, s)
 }
 
+func (m *Store) LockSnapshotRequests(ctx context.Context, reqs []*snapshot.Request) (func(), error) {
+	if m.LockSnapshotRequestsFn == nil {
+		return func() {}, nil
+	}
+	return m.LockSnapshotRequestsFn(ctx, reqs)
+}
+
+func (m *Store) ResumeSnapshotRequest(ctx context.Context, req *snapshot.Request) (bool, error) {
+	return m.ResumeSnapshotRequestFn(ctx, req)
+}
+
 func (m *Store) GetSnapshotRequestsByStatus(ctx context.Context, status snapshot.Status) ([]*snapshot.Request, error) {
 	return m.GetSnapshotRequestsByStatusFn(ctx, status)
 }
diff --git a/pkg/snapshot/store/postgres/integration/pg_snapshot_store_recovery_integration_test.go b/pkg/snapshot/store/postgres/integration/pg_snapshot_store_recovery_integration_test.go
new file mode 100644
index 000000000..282886a4e
--- /dev/null
+++ b/pkg/snapshot/store/postgres/integration/pg_snapshot_store_recovery_integration_test.go
@@ -0,0 +1,255 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package integration
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	pglib "github.com/xataio/pgstream/internal/postgres"
+	"github.com/xataio/pgstream/pkg/snapshot"
+	"github.com/xataio/pgstream/pkg/snapshot/generator"
+	"github.com/xataio/pgstream/pkg/snapshot/store"
+	pgstore "github.com/xataio/pgstream/pkg/snapshot/store/postgres"
+)
+
+// Test_SnapshotStore_LockSnapshotRequests checks that only one holder can lock
+// a schema and table list at a time, whatever the request mode, and that the
+// lock is free again once released.
+func Test_SnapshotStore_LockSnapshotRequests(t *testing.T) {
+	if os.Getenv("PGSTREAM_INTEGRATION_TESTS") == "" {
+		t.Skip("skipping integration test...")
+	}
+
+	ctx := context.Background()
+	s, err := pgstore.New(ctx, sharedPGURL)
+	require.NoError(t, err)
+	defer s.Close()
+
+	dataReq := &snapshot.Request{Schema: "lock_schema", Tables: []string{"t1"}, Mode: snapshot.RequestModeData}
+	schemaOnlyReq := &snapshot.Request{Schema: "lock_schema", Tables: []string{"t1"}, Mode: snapshot.RequestModeSchemaOnly}
+	otherReq := &snapshot.Request{Schema: "lock_schema", Tables: []string{"t2"}}
+
+	release, err := s.LockSnapshotRequests(ctx, []*snapshot.Request{dataReq})
+	require.NoError(t, err)
+
+	// the unique index ignores mode, so the lock does too
+	_, err = s.LockSnapshotRequests(ctx, []*snapshot.Request{schemaOnlyReq})
+	require.ErrorIs(t, err, store.ErrSnapshotRequestLocked)
+
+	// a different table list is a different scope
+	releaseOther, err := s.LockSnapshotRequests(ctx, []*snapshot.Request{otherReq})
+	require.NoError(t, err)
+	releaseOther()
+
+	release()
+	release, err = s.LockSnapshotRequests(ctx, []*snapshot.Request{schemaOnlyReq})
+	require.NoError(t, err)
+	release()
+}
+
+// Test_SnapshotStore_LockReleasedWhenHolderDies checks that the server frees
+// the lock of a run whose connection is gone, as after a kill -9, so a later
+// run can recover its request.
+func Test_SnapshotStore_LockReleasedWhenHolderDies(t *testing.T) {
+	if os.Getenv("PGSTREAM_INTEGRATION_TESTS") == "" {
+		t.Skip("skipping integration test...")
+	}
+
+	ctx := context.Background()
+	s, err := pgstore.New(ctx, sharedPGURL)
+	require.NoError(t, err)
+	defer s.Close()
+
+	req := &snapshot.Request{Schema: "lock_dies_schema", Tables: []string{"t1"}}
+	release, err := s.LockSnapshotRequests(ctx, []*snapshot.Request{req})
+	require.NoError(t, err)
+	defer release()
+
+	_, err = s.LockSnapshotRequests(ctx, []*snapshot.Request{req})
+	require.ErrorIs(t, err, store.ErrSnapshotRequestLocked)
+
+	// end the holder's session from outside, like a dead process would
+	conn, err := pglib.NewConn(ctx, sharedPGURL)
+	require.NoError(t, err)
+	defer conn.Close(ctx)
+	var terminated int
+	err = conn.QueryRow(ctx, []any{&terminated}, fmt.Sprintf(`SELECT count(pg_terminate_backend(l.pid))::int FROM pg_locks l
+	WHERE l.locktype = 'advisory' AND l.granted
+	AND ((l.classid::bigint << 32) | l.objid::bigint) = hashtextextended($1::text || '/' || %s.snapshot_table_names_hash($2::text[]), 0)`, store.SchemaName),
+		req.Schema, req.Tables)
+	require.NoError(t, err)
+	require.Equal(t, 1, terminated)
+
+	require.Eventually(t, func() bool {
+		release, err := s.LockSnapshotRequests(ctx, []*snapshot.Request{req})
+		if err != nil {
+			return false
+		}
+		release()
+		return true
+	}, 10*time.Second, 100*time.Millisecond)
+}
+
+// Test_SnapshotStore_ResumeSnapshotRequest checks the take over of a leftover
+// request: it matches the unique index key whatever the mode, records the new
+// mode so the completion update finds the row, and reports when there is
+// nothing left to take over.
+func Test_SnapshotStore_ResumeSnapshotRequest(t *testing.T) {
+	if os.Getenv("PGSTREAM_INTEGRATION_TESTS") == "" {
+		t.Skip("skipping integration test...")
+	}
+
+	ctx := context.Background()
+	clearRequests(t, "resume_schema")
+
+	s, err := pgstore.New(ctx, sharedPGURL)
+	require.NoError(t, err)
+	defer s.Close()
+
+	// a schema-only run left its request behind
+	leftover := &snapshot.Request{Schema: "resume_schema", Tables: []string{"t1"}, Mode: snapshot.RequestModeSchemaOnly}
+	require.NoError(t, s.CreateSnapshotRequest(ctx, leftover))
+
+	// a data run for the same tables conflicts on the index, which ignores mode
+	req := &snapshot.Request{Schema: "resume_schema", Tables: []string{"t1"}, Mode: snapshot.RequestModeData}
+	require.ErrorIs(t, s.CreateSnapshotRequest(ctx, req), store.ErrNonCompletedRequestExists)
+
+	resumed, err := s.ResumeSnapshotRequest(ctx, req)
+	require.NoError(t, err)
+	require.True(t, resumed)
+	requireRequests(t, s, "resume_schema", []snapshot.Status{snapshot.StatusInProgress}, snapshot.RequestModeData)
+
+	// the completion update is keyed on mode and must find the row
+	req.MarkCompleted(req.Schema, nil)
+	require.NoError(t, s.UpdateSnapshotRequest(ctx, req))
+	requireRequests(t, s, "resume_schema", []snapshot.Status{snapshot.StatusCompleted}, snapshot.RequestModeData)
+
+	// nothing non-completed is left, so there is nothing to take over
+	resumed, err = s.ResumeSnapshotRequest(ctx, req)
+	require.NoError(t, err)
+	require.False(t, resumed)
+}
+
+// Test_SnapshotRecorder_ConcurrentRuns runs the recorder against real postgres
+// with repeatable snapshots. While a slow run is copying, two retries started
+// together are refused and leave its request alone, so neither can complete
+// it early and the slow run's own completion updates its own row. A run that
+// died (its request is left in progress and its lock is gone) is recovered.
+func Test_SnapshotRecorder_ConcurrentRuns(t *testing.T) {
+	if os.Getenv("PGSTREAM_INTEGRATION_TESTS") == "" {
+		t.Skip("skipping integration test...")
+	}
+
+	ctx := context.Background()
+	const schema = "concurrent_schema"
+	clearRequests(t, schema)
+
+	newRecorder := func(gen generator.SnapshotGenerator) *generator.SnapshotRecorder {
+		s, err := pgstore.New(ctx, sharedPGURL)
+		require.NoError(t, err)
+		return generator.NewSnapshotRecorder(&generator.Config{RepeatableSnapshots: true}, s, gen)
+	}
+	newSnapshot := func() *snapshot.Snapshot {
+		return &snapshot.Snapshot{SchemaTables: map[string][]string{schema: {"t1"}}}
+	}
+
+	s, err := pgstore.New(ctx, sharedPGURL)
+	require.NoError(t, err)
+	defer s.Close()
+
+	// slow writer: blocks until told to finish
+	started := make(chan struct{})
+	finish := make(chan struct{})
+	slow := newRecorder(&blockingGenerator{started: started, finish: finish})
+	defer slow.Close()
+	slowErr := make(chan error, 1)
+	go func() { slowErr <- slow.CreateSnapshot(ctx, newSnapshot()) }()
+	<-started
+	requireRequests(t, s, schema, []snapshot.Status{snapshot.StatusInProgress}, snapshot.RequestModeData)
+
+	// two retries at the same time while the slow run is still copying. The
+	// recorders are built first, since building a store runs its table setup.
+	retryGens := []*recordingGenerator{{}, {}}
+	retries := make([]*generator.SnapshotRecorder, 0, len(retryGens))
+	for _, gen := range retryGens {
+		r := newRecorder(gen)
+		defer r.Close()
+		retries = append(retries, r)
+	}
+	var wg sync.WaitGroup
+	retryErrs := make([]error, len(retries))
+	for i, r := range retries {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			retryErrs[i] = r.CreateSnapshot(ctx, newSnapshot())
+		}()
+	}
+	wg.Wait()
+	for i := range retries {
+		require.ErrorIs(t, retryErrs[i], store.ErrSnapshotRequestLocked)
+		require.False(t, retryGens[i].called, "a refused retry must not copy")
+	}
+	// the slow run's request is untouched by the refused retries
+	requireRequests(t, s, schema, []snapshot.Status{snapshot.StatusInProgress}, snapshot.RequestModeData)
+
+	close(finish)
+	require.NoError(t, <-slowErr)
+	requireRequests(t, s, schema, []snapshot.Status{snapshot.StatusCompleted}, snapshot.RequestModeData)
+
+	// a run that died: its request stays in progress and nobody holds its lock
+	dead := &snapshot.Request{Schema: schema, Tables: []string{"t1"}, Mode: snapshot.RequestModeData}
+	require.NoError(t, s.CreateSnapshotRequest(ctx, dead))
+	dead.MarkInProgress()
+	require.NoError(t, s.UpdateSnapshotRequest(ctx, dead))
+
+	gen := &recordingGenerator{}
+	recovery := newRecorder(gen)
+	defer recovery.Close()
+	require.NoError(t, recovery.CreateSnapshot(ctx, newSnapshot()))
+	require.True(t, gen.called)
+	requireRequests(t, s, schema, []snapshot.Status{snapshot.StatusCompleted, snapshot.StatusCompleted}, snapshot.RequestModeData)
+}
+
+type blockingGenerator struct {
+	started chan struct{}
+	finish  chan struct{}
+}
+
+func (g *blockingGenerator) CreateSnapshot(ctx context.Context, _ *snapshot.Snapshot) error {
+	close(g.started)
+	<-g.finish
+	return nil
+}
+
+func (g *blockingGenerator) Close() error { return nil }
+
+type recordingGenerator struct {
+	called bool
+}
+
+func (g *recordingGenerator) CreateSnapshot(context.Context, *snapshot.Snapshot) error {
+	g.called = true
+	return nil
+}
+
+func (g *recordingGenerator) Close() error { return nil }
+
+func requireRequests(t *testing.T, s *pgstore.Store, schema string, wantStatuses []snapshot.Status, wantMode snapshot.RequestMode) {
+	t.Helper()
+
+	reqs, err := s.GetSnapshotRequestsBySchema(context.Background(), schema)
+	require.NoError(t, err)
+	statuses := make([]snapshot.Status, 0, len(reqs))
+	for _, req := range reqs {
+		statuses = append(statuses, req.Status)
+		require.Equal(t, wantMode, req.GetMode())
+	}
+	require.Equal(t, wantStatuses, statuses)
+}
diff --git a/pkg/snapshot/store/postgres/pg_snapshot_store.go b/pkg/snapshot/store/postgres/pg_snapshot_store.go
index 7a0fb94b8..ca0981f71 100644
--- a/pkg/snapshot/store/postgres/pg_snapshot_store.go
+++ b/pkg/snapshot/store/postgres/pg_snapshot_store.go
@@ -4,7 +4,10 @@ package postgres
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"slices"
+	"strings"
 
 	"github.com/lib/pq"
 	"github.com/xataio/pgstream/internal/postgres"
@@ -14,9 +17,17 @@ import (
 
 type Store struct {
 	conn postgres.Querier
+	// lockConnBuilder opens the dedicated connection that holds the request
+	// locks. Session level advisory locks belong to one connection, so they
+	// can't be taken through the pool.
+	lockConnBuilder func(context.Context) (postgres.Querier, error)
 }
 
-const queryLimit = 1000
+const (
+	queryLimit = 1000
+
+	uniqueIndexName = "schema_table_status_hash_unique_index"
+)
 
 func New(ctx context.Context, url string) (*Store, error) {
 	conn, err := postgres.NewConnPool(ctx, url)
@@ -26,6 +37,9 @@ func New(ctx context.Context, url string) (*Store, error) {
 
 	s := &Store{
 		conn: conn,
+		lockConnBuilder: func(ctx context.Context) (postgres.Querier, error) {
+			return postgres.NewConn(ctx, url)
+		},
 	}
 
 	// create snapshots table if it doesn't exist
@@ -45,12 +59,79 @@ func (s *Store) CreateSnapshotRequest(ctx context.Context, req *snapshot.Request
 	VALUES($1, $2, now(), now(),'requested', $3)`, snapshotsTable())
 	_, err := s.conn.Exec(ctx, query, req.Schema, pq.StringArray(req.Tables), req.GetMode())
 	if err != nil {
+		if isNonCompletedRequestConflict(err) {
+			return fmt.Errorf("error creating snapshot request: %w: %w", store.ErrNonCompletedRequestExists, err)
+		}
 		return fmt.Errorf("error creating snapshot request: %w", err)
 	}
 
 	return nil
 }
 
+// isNonCompletedRequestConflict reports whether err is the unique violation
+// on the index that allows one non-completed request per schema and table
+// list. Other integrity errors (check or not null constraints) are not.
+func isNonCompletedRequestConflict(err error) bool {
+	constraintErr := &postgres.ErrConstraintViolation{}
+	return errors.As(err, &constraintErr) && strings.Contains(constraintErr.Details, uniqueIndexName)
+}
+
+// LockSnapshotRequests takes a session level advisory lock per request, keyed
+// like the unique index on the schema name and the table names hash (mode is
+// not part of the key). The locks live on a dedicated connection, so they are
+// released when release closes it, or by the server when the process dies and
+// its connection drops.
+func (s *Store) LockSnapshotRequests(ctx context.Context, reqs []*snapshot.Request) (func(), error) {
+	conn, err := s.lockConnBuilder(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("opening snapshot request lock connection: %w", err)
+	}
+	release := func() {
+		conn.Close(context.Background())
+	}
+
+	// take the locks in a fixed order so two runs over the same scopes
+	// contend on the same first lock
+	sorted := slices.Clone(reqs)
+	slices.SortFunc(sorted, func(a, b *snapshot.Request) int {
+		return strings.Compare(lockKey(a), lockKey(b))
+	})
+
+	query := fmt.Sprintf(`SELECT pg_try_advisory_lock(hashtextextended($1::text || '/' || %s.snapshot_table_names_hash($2::text[]), 0))`, store.SchemaName)
+	for _, req := range sorted {
+		var locked bool
+		if err := conn.QueryRow(ctx, []any{&locked}, query, req.Schema, pq.StringArray(req.Tables)); err != nil {
+			release()
+			return nil, fmt.Errorf("locking snapshot request for schema %q: %w", req.Schema, err)
+		}
+		if !locked {
+			release()
+			return nil, fmt.Errorf("schema %q tables %v: %w", req.Schema, req.Tables, store.ErrSnapshotRequestLocked)
+		}
+	}
+
+	return release, nil
+}
+
+func lockKey(req *snapshot.Request) string {
+	return req.Schema + "/" + strings.Join(req.Tables, ",")
+}
+
+// ResumeSnapshotRequest takes over the non-completed request matching the
+// unique index key (schema name and table names, any mode), marks it in
+// progress and records the mode on input, so later updates keyed on mode
+// match it.
+func (s *Store) ResumeSnapshotRequest(ctx context.Context, req *snapshot.Request) (bool, error) {
+	query := fmt.Sprintf(`UPDATE %s SET status = 'in progress', mode = $1, errors = NULL, updated_at = now()
+	WHERE schema_name = $2 and table_names = $3 and status != 'completed'`, snapshotsTable())
+	tag, err := s.conn.Exec(ctx, query, req.GetMode(), req.Schema, pq.StringArray(req.Tables))
+	if err != nil {
+		return false, fmt.Errorf("error resuming snapshot request: %w", err)
+	}
+
+	return tag.RowsAffected() > 0, nil
+}
+
 func (s *Store) UpdateSnapshotRequest(ctx context.Context, req *snapshot.Request) error {
 	query := fmt.Sprintf(`UPDATE %s SET status = $1, errors = $2, updated_at = now()
 	WHERE schema_name = $3 and table_names = $4 and mode = $5 and status != 'completed'`, snapshotsTable())
diff --git a/pkg/snapshot/store/postgres/pg_snapshot_store_test.go b/pkg/snapshot/store/postgres/pg_snapshot_store_test.go
index 5ac6039d1..1c38d1602 100644
--- a/pkg/snapshot/store/postgres/pg_snapshot_store_test.go
+++ b/pkg/snapshot/store/postgres/pg_snapshot_store_test.go
@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/lib/pq"
 	"github.com/stretchr/testify/require"
 	pglib "github.com/xataio/pgstream/internal/postgres"
@@ -585,3 +586,160 @@ func TestStore_createTable(t *testing.T) {
 		})
 	}
 }
+
+func TestStore_CreateSnapshotRequest_conflict(t *testing.T) {
+	t.Parallel()
+
+	req := &snapshot.Request{Schema: "public", Tables: []string{"t1"}}
+
+	tests := []struct {
+		name      string
+		execErr   error
+		wantStale bool
+	}{
+		{
+			name:      "unique violation on the non-completed request index",
+			execErr:   &pglib.ErrConstraintViolation{Details: `duplicate key value violates unique constraint "schema_table_status_hash_unique_index"`},
+			wantStale: true,
+		},
+		{
+			name:      "check constraint violation",
+			execErr:   &pglib.ErrConstraintViolation{Details: `new row for relation "snapshot_requests" violates check constraint "snapshot_requests_mode_check"`},
+			wantStale: false,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			s := &Store{
+				conn: &postgresmocks.Querier{
+					ExecFn: func(ctx context.Context, _ uint, _ string, _ ...any) (pglib.CommandTag, error) {
+						return pglib.CommandTag{}, tc.execErr
+					},
+				},
+			}
+			err := s.CreateSnapshotRequest(context.Background(), req)
+			require.ErrorIs(t, err, tc.execErr)
+			require.Equal(t, tc.wantStale, errors.Is(err, store.ErrNonCompletedRequestExists))
+		})
+	}
+}
+
+func TestStore_ResumeSnapshotRequest(t *testing.T) {
+	t.Parallel()
+
+	req := &snapshot.Request{Schema: "public", Tables: []string{"t1"}, Mode: snapshot.RequestModeSchemaOnly}
+	errTest := errors.New("oh noes")
+
+	tests := []struct {
+		name    string
+		tag     string
+		execErr error
+
+		wantResumed bool
+		wantErr     error
+	}{
+		{name: "takes over the leftover request", tag: "UPDATE 1", wantResumed: true},
+		{name: "no leftover request", tag: "UPDATE 0", wantResumed: false},
+		{name: "error", execErr: errTest, wantErr: errTest},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			s := &Store{
+				conn: &postgresmocks.Querier{
+					ExecFn: func(ctx context.Context, _ uint, q string, a ...any) (pglib.CommandTag, error) {
+						wantQuery := fmt.Sprintf(`UPDATE %s SET status = 'in progress', mode = $1, errors = NULL, updated_at = now()
+	WHERE schema_name = $2 and table_names = $3 and status != 'completed'`, snapshotsTable())
+						require.Equal(t, wantQuery, q)
+						require.Equal(t, []any{snapshot.RequestModeSchemaOnly, "public", pq.StringArray{"t1"}}, a)
+						return pglib.CommandTag{CommandTag: pgconn.NewCommandTag(tc.tag)}, tc.execErr
+					},
+				},
+			}
+			resumed, err := s.ResumeSnapshotRequest(context.Background(), req)
+			require.ErrorIs(t, err, tc.wantErr)
+			require.Equal(t, tc.wantResumed, resumed)
+		})
+	}
+}
+
+func TestStore_LockSnapshotRequests(t *testing.T) {
+	t.Parallel()
+
+	reqs := []*snapshot.Request{
+		{Schema: "b", Tables: []string{"t1"}},
+		{Schema: "a", Tables: []string{"t1"}},
+	}
+	errTest := errors.New("oh noes")
+
+	tests := []struct {
+		name      string
+		locked    []bool
+		queryErr  error
+		connErr   error
+		wantErr   error
+		wantCalls int
+	}{
+		{name: "all locks taken", locked: []bool{true, true}, wantCalls: 2},
+		{name: "second lock held by another run", locked: []bool{true, false}, wantErr: store.ErrSnapshotRequestLocked, wantCalls: 2},
+		{name: "lock query fails", queryErr: errTest, wantErr: errTest, wantCalls: 1},
+		{name: "lock connection fails", connErr: errTest, wantErr: errTest},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			calls, closes := 0, 0
+			var schemas []string
+			conn := &postgresmocks.Querier{
+				QueryRowFn: func(ctx context.Context, dest []any, q string, a ...any) error {
+					calls++
+					schema, ok := a[0].(string)
+					require.True(t, ok)
+					schemas = append(schemas, schema)
+					if tc.queryErr != nil {
+						return tc.queryErr
+					}
+					locked, ok := dest[0].(*bool)
+					require.True(t, ok)
+					*locked = tc.locked[calls-1]
+					return nil
+				},
+				CloseFn: func(context.Context) error {
+					closes++
+					return nil
+				},
+			}
+			s := &Store{
+				lockConnBuilder: func(context.Context) (pglib.Querier, error) {
+					if tc.connErr != nil {
+						return nil, tc.connErr
+					}
+					return conn, nil
+				},
+			}
+
+			release, err := s.LockSnapshotRequests(context.Background(), reqs)
+			require.ErrorIs(t, err, tc.wantErr)
+			require.Equal(t, tc.wantCalls, calls)
+			if tc.wantCalls > 0 {
+				// locks are taken in a fixed order
+				require.Equal(t, "a", schemas[0])
+			}
+			if err != nil {
+				require.Nil(t, release)
+				require.Equal(t, min(tc.wantCalls, 1), closes)
+				return
+			}
+			require.Equal(t, 0, closes)
+			release()
+			require.Equal(t, 1, closes)
+		})
+	}
+}
diff --git a/pkg/snapshot/store/snapshot_store.go b/pkg/snapshot/store/snapshot_store.go
index 4f31bd113..f03d63f0e 100644
--- a/pkg/snapshot/store/snapshot_store.go
+++ b/pkg/snapshot/store/snapshot_store.go
@@ -4,18 +4,37 @@ package store
 
 import (
 	"context"
+	"errors"
 
 	"github.com/xataio/pgstream/pkg/snapshot"
 )
 
 type Store interface {
+	// CreateSnapshotRequest records a new request. It returns an error
+	// wrapping ErrNonCompletedRequestExists when a non-completed request for
+	// the same schema and table list is already recorded.
 	CreateSnapshotRequest(context.Context, *snapshot.Request) error
 	UpdateSnapshotRequest(context.Context, *snapshot.Request) error
+	// LockSnapshotRequests takes a lock per request on its schema and table
+	// list, held until release is called or the process exits, so only one
+	// run can work on a given scope at a time. It returns an error wrapping
+	// ErrSnapshotRequestLocked when another run holds one of the locks.
+	LockSnapshotRequests(context.Context, []*snapshot.Request) (release func(), err error)
+	// ResumeSnapshotRequest takes over the non-completed request recorded for
+	// the same schema and table list as the one on input, whatever its mode,
+	// and marks it in progress with the mode on input. It reports whether a
+	// request was taken over. The caller must hold the request lock.
+	ResumeSnapshotRequest(context.Context, *snapshot.Request) (bool, error)
 	GetSnapshotRequestsByStatus(ctx context.Context, status snapshot.Status) ([]*snapshot.Request, error)
 	GetSnapshotRequestsBySchema(ctx context.Context, schema string) ([]*snapshot.Request, error)
 	Close() error
 }
 
+var (
+	ErrNonCompletedRequestExists = errors.New("a non-completed snapshot request already exists")
+	ErrSnapshotRequestLocked     = errors.New("snapshot request locked by another run")
+)
+
 const (
 	SchemaName = "pgstream"
 	TableName  = "snapshot_requests"