diff --git a/docs/design/single-run-lock.md b/docs/design/single-run-lock.md
new file mode 100644
index 0000000..bf0e9a6
--- /dev/null
+++ b/docs/design/single-run-lock.md
@@ -0,0 +1,224 @@
+# Design: single-run lock per ingestion job
+
+Status: **draft for review** · Area: `application` / `infra.store`
+
+## Problem
+
+ingestify assumes an `(IngestionPlan, Selector)` job runs in **one process at a time**.
+Nothing enforces it. A scheduler (e.g. Cloud Run Job + Cloud Scheduler) or a restart of a
+crashed container can start a second execution of the same job while the first is still
+running — Cloud Run Jobs have no "max 1 execution" setting.
+
+Two runs of the same job identity produce **undefined results**: they discover and write
+the same datasets concurrently, and stateful sources double their work. Observed with a
+paid, task-based async source: each run took an independent server-side snapshot at startup,
+missed the other's just-submitted tasks, and **re-submitted (paid for) the same work**. The
+`ingestion_job_summary` table also accumulated rows stuck in `RUNNING` from crashed
+overlapping runs.
+
+Because concurrent same-job execution is never well-defined, this is **not a knob** to
+expose — it is an **invariant** the framework should uphold.
+
+## Decision
+
+ingestify enforces, best-effort, that **one job identity never runs in two processes at
+once**. Always on, not configurable. Where the metadata store can provide a cross-process
+lock (Postgres) it is enforced hard; where it cannot (SQLite / local single-process) it is
+a documented no-op.
+
+Not a semaphore, not `max_concurrent_runs: N` — a plain **mutex per job identity**. Parallel
+partitions are still possible the correct way: give them **different selectors** → different
+identities → different locks.
+
+### Naming
+
+Call it a **run lock** / single-run guarantee. Do **not** call it "concurrency":
+`Source.max_concurrency` already exists and means something unrelated — the number of
+worker processes for tasks *within* one job (`Loader.run` → `TaskExecutor(processes=...)`,
+`loader.py:276`).
+
+## Job identity (lock scope)
+
+`(source.name, dataset_type, selector.key)` — exactly the merge key the loader already uses
+to guarantee one dataset per combination (`loader.py:193-197`). Serialised to a stable
+string:
+
+```
+job_key = f"{source.name}:{dataset_type}:{selector.key}"
+```
+
+Different selector → different `job_key` → runs in parallel. Same identity → competes for
+the one lock.
+
+## Mechanism — backend session lock (mutex)
+
+The metadata store already owns a SQLAlchemy engine (`SqlAlchemySessionProvider.engine`,
+`repository.py:95`) and already branches on dialect (`repository.py:171`). Both supported
+server databases provide a **session-scoped** lock — held for the life of one connection,
+released automatically when it ends:
+
+- **PostgreSQL** — `pg_try_advisory_lock(:key)` / `pg_advisory_unlock(:key)`, with
+  `key = signed_bigint(hash64(job_key))` (advisory locks take a bigint).
+- **MySQL** — `GET_LOCK(:name, 0)` / `RELEASE_LOCK(:name)`, with `name = hex(hash(job_key))`
+  (user-level lock names are ≤ 64 chars, so hash to a bounded string). `GET_LOCK(…, 0)` =
+  try once, non-blocking: `1` = obtained, `0` = held by another session.
+
+Same flow for both, on a **dedicated** connection (not the `scoped_session`, which is
+reused/closed):
+
+- try the lock → obtained → keep the connection open for the whole run.
+- not obtained → another process holds it → the caller skips.
+- release explicitly at the end + close the connection; on crash the connection drops and
+  the server **releases the lock automatically** — no stale locks, no cleanup job.
+
+Why a session lock: atomic (no "list executions then decide" race), self-healing (dies with
+the connection), zero extra infra (reuses the metadata DB).
+
+### Lock lifetime — released the moment the process stops
+
+The lock lives exactly as long as its **connection/session**, nothing more. On any process
+stop — clean exit, exception, `SIGKILL`, OOM-kill, container recycle — the socket closes and
+the server (Postgres advisory lock or MySQL `GET_LOCK`, both session-scoped) releases the
+lock **immediately**. No TTL, no heartbeat, no lease renewal, no cleanup job. That is the
+whole reason to prefer this over a "lock row with an expiry" table, which would either block
+forever after a crash or need a reaper.
+
+One caveat, stated honestly: release depends on the DB *seeing* the connection close. A hard
+network partition that leaves the socket half-open (client vanishes without a FIN) is the
+single case where release waits on TCP keepalive instead of being instant — bounded by the
+server's keepalive settings, never indefinite. For a job process that simply ends, the
+socket closes and release is prompt. Consequence for the implementation: the dedicated lock
+connection must **not** silently auto-reconnect (a reconnect is a *new* session, without the
+lock) — a dropped lock connection is fatal to the run.
+
+## API surface (grounded in the current code)
+
+**`DatasetRepository`** (`domain/models/dataset/dataset_repository.py`, ABC) — new method:
+
+```python
+@abstractmethod
+def acquire_run_lock(self, job_key: str) -> "RunLock | None":
+    """Return a held RunLock, or None if another process holds this job_key.
+    Stores without cross-process locking return an always-acquired no-op lock."""
+```
+
+**`RunLock`** — tiny handle:
+
+```python
+class RunLock:
+    def release(self) -> None: ...
+```
+
+**`SqlAlchemyDatasetRepository`** (`infra/store/dataset/sqlalchemy/repository.py:208`):
+
+```python
+def acquire_run_lock(self, job_key: str) -> RunLock | None:
+    dialect = self.dialect.name
+    conn = self.session_provider.engine.connect()   # dedicated, outside the scoped_session
+    if dialect == "postgresql":
+        key = _signed_bigint(job_key)               # advisory locks take a bigint
+        got = conn.execute(text("SELECT pg_try_advisory_lock(:k)"), {"k": key}).scalar()
+        acquire, unlock = bool(got), text("SELECT pg_advisory_unlock(:k)")
+    elif dialect == "mysql":
+        name = _lock_name(job_key)                   # hex hash, <= 64 chars
+        got = conn.execute(text("SELECT GET_LOCK(:n, 0)"), {"n": name}).scalar()
+        acquire, unlock = (got == 1), text("SELECT RELEASE_LOCK(:n)")
+    else:
+        conn.close()
+        return _NoopRunLock()                        # no cross-process lock (e.g. SQLite)
+    if not acquire:
+        conn.close()
+        return None
+    return _SessionRunLock(conn, unlock, params)     # holds conn; release() unlocks + closes
+```
+
+**`DatasetStore`** (`application/dataset_store.py:112`) delegates, exactly like
+`save_ingestion_job_summary` (`dataset_store.py:194`):
+
+```python
+def acquire_run_lock(self, job_key: str) -> RunLock | None:
+    return self.dataset_repository.acquire_run_lock(job_key)
+```
+
+**`IngestionJob.execute`** (`domain/models/ingestion/ingestion_job.py:283`) — acquire at the
+very top, before the `RUNNING` summary is created/persisted (`:290`):
+
+```python
+job_key = f"{self.ingestion_plan.source.name}:{self.ingestion_plan.dataset_type}:{self.selector.key}"
+lock = store.acquire_run_lock(job_key)
+if lock is None:
+    summary = IngestionJobSummary.new(ingestion_job=self)
+    summary.set_skipped(reason="another run holds the lock")
+    store.save_ingestion_job_summary(summary)
+    yield summary
+    return
+try:
+    ...  # existing execute body (state=RUNNING, submit/collect, set_finished)
+finally:
+    lock.release()
+```
+
+`execute` is a generator fully consumed by `Loader.run`, so the lock is held across the
+whole run and released in `finally` on normal completion, exception, or `GeneratorExit`.
+
+**`IngestionJobSummary`** — add a `SKIPPED` state + `set_skipped(reason)` (sits alongside
+the existing `set_finished()`), so a skipped run is observable and distinct from RUNNING /
+FINISHED. A skipped job does **not** create a RUNNING row.
+
+## Behaviour on conflict
+
+Skip: record a `SKIPPED` summary, do not run, let `Loader.run` continue to the next job.
+The process exits **0** — a scheduler must not see a skip as a failure or retry it.
+
+## Edge cases & failure modes
+
+- **Crash mid-run** → connection drops → lock frees → next run proceeds. (Summary row stays
+  RUNNING until Phase 2; correctness unaffected.)
+- **Lock connection lost but process alive** (DB blip) → the run is now unprotected. Treat a
+  dead lock connection as fatal: abort rather than continue.
+- **Store without a session lock** (e.g. SQLite) → no cross-process guarantee (no-op,
+  documented). Local runs are single-process anyway.
+- **Multi-primary cluster** (Galera / multiple write nodes) → both `pg_advisory_lock` and
+  MySQL `GET_LOCK` are *per node*, not cluster-wide, so two runs on two write nodes would not
+  see each other. The common single-primary deployment is fully covered; documented as a
+  limitation for multi-primary setups.
+- **hash collision** between two identities → they'd share one lock (over-restrict, never
+  under-restrict). Safe direction; negligible probability. Documented, not mitigated.
+
+## Testing
+
+- Two connections to a test **Postgres or MySQL**: first `acquire_run_lock(k)` holds; second
+  returns `None`; after `release()` the second succeeds. Same test parametrised over both
+  dialects (skips the one not available).
+- Crash: close the holder connection → next acquire succeeds (auto-release).
+- Store without a session lock (SQLite) → always grants (no-op contract).
+- Integration: two overlapping `IngestionJob.execute` for the same identity → one runs, the
+  other yields a `SKIPPED` summary; different selectors → both run.
+
+## Phase 2 (separate) — reconcile stale RUNNING
+
+Advisory locks free on crash, but `ingestion_job_summary` rows stay `RUNNING`. On acquiring
+a lock, a `RUNNING` summary for the same `job_key` whose advisory lock is **not** currently
+held (checked via `pg_locks`) is orphaned → mark `ABANDONED`/`FAILED`. Observability
+cleanup, not correctness; keep out of phase 1.
+
+## Backwards compatibility
+
+Always-on is safe: concurrent same-job execution was already undefined, so enforcing it can
+only prevent breakage. A run that previously "worked" under overlap by luck now gets a
+clean `SKIPPED` instead of racing. Nothing to configure, nothing removed.
+
+## Rollout
+
+1. Implement (repository method + Postgres impl + `RunLock` + `DatasetStore` delegation +
+   `IngestionJob.execute` integration + `SKIPPED` summary state + tests).
+2. Release ingestify (version bump).
+3. Consumers bump the dep; a scheduler that would otherwise start an overlapping run now
+   self-skips instead of needing a manual pause.
+
+## Open questions
+
+1. `SKIPPED` as a first-class summary state (proposed) vs. just not persisting anything for
+   a skipped run. First-class is better for observability.
+2. Acquire inside `IngestionJob.execute` (proposed — owns the lifecycle) vs. in `Loader.run`
+   around it. Either works; `execute` keeps it next to the RUNNING/summary logic.
diff --git a/ingestify/application/dataset_store.py b/ingestify/application/dataset_store.py
index 8aa0156..451564c 100644
--- a/ingestify/application/dataset_store.py
+++ b/ingestify/application/dataset_store.py
@@ -194,6 +194,11 @@ def with_file_cache(self):
     def save_ingestion_job_summary(self, ingestion_job_summary):
         self.dataset_repository.save_ingestion_job_summary(ingestion_job_summary)
 
+    def acquire_run_lock(self, job_key: str):
+        """Single-run lock for one job identity (see docs/design/single-run-lock.md).
+        Returns a held RunLock, or None if another process already holds it."""
+        return self.dataset_repository.acquire_run_lock(job_key)
+
     def get_dataset_last_modified_at_map(
         self, provider: str, dataset_type: str
     ) -> "DatasetLastModifiedAtMap":
diff --git a/ingestify/domain/models/dataset/dataset_repository.py b/ingestify/domain/models/dataset/dataset_repository.py
index eb39cdd..27e8f2a 100644
--- a/ingestify/domain/models/dataset/dataset_repository.py
+++ b/ingestify/domain/models/dataset/dataset_repository.py
@@ -8,7 +8,26 @@
 from .selector import Selector
 
 
+class RunLock:
+    """Handle for a held single-run lock (one per job identity). ``release()`` frees it;
+    it is also released automatically when the owning DB connection/process ends."""
+
+    def release(self) -> None:
+        pass
+
+
+class NoopRunLock(RunLock):
+    """Always-granted lock for stores without cross-process locking (e.g. SQLite)."""
+
+
 class DatasetRepository(ABC):
+    def acquire_run_lock(self, job_key: str) -> Optional[RunLock]:
+        """Best-effort single-run lock for one job identity (design:
+        docs/design/single-run-lock.md). Return a held ``RunLock``, or ``None`` if another
+        process already holds it. Stores without cross-process locking return an
+        always-granted no-op lock, so a lone local process is never blocked."""
+        return NoopRunLock()
+
     @abstractmethod
     def get_dataset_collection(
         self,
diff --git a/ingestify/domain/models/ingestion/ingestion_job.py b/ingestify/domain/models/ingestion/ingestion_job.py
index dadfa61..f67bce1 100644
--- a/ingestify/domain/models/ingestion/ingestion_job.py
+++ b/ingestify/domain/models/ingestion/ingestion_job.py
@@ -280,11 +280,40 @@ def _save_progress(
         ingestion_job_summary.recount()
         store.save_ingestion_job_summary(ingestion_job_summary)
 
+    def _job_key(self) -> str:
+        # One lock per (source, dataset_type, selector) -- the same identity the loader
+        # uses to guarantee one dataset per combination.
+        return (
+            f"{self.ingestion_plan.source.name}:{self.ingestion_plan.dataset_type}"
+            f":{self.selector.key}"
+        )
+
     def execute(
         self,
         store: DatasetStore,
         task_executor: TaskExecutor,
         last_modified_at_map: Optional[DatasetLastModifiedAtMap] = None,
+    ) -> Iterator[IngestionJobSummary]:
+        # Single-run guard: one job identity must never run in two processes at once
+        # (design: docs/design/single-run-lock.md). The lock is a session-scoped DB lock,
+        # released the instant this process ends. If another run holds it, skip cleanly.
+        run_lock = store.acquire_run_lock(self._job_key())
+        if run_lock is None:
+            logger.info("Skipping %s: another run holds the lock", self._job_key())
+            summary = IngestionJobSummary.new(ingestion_job=self)
+            summary.set_skipped()
+            yield summary  # Loader persists every yielded summary
+            return
+        try:
+            yield from self._execute_locked(store, task_executor, last_modified_at_map)
+        finally:
+            run_lock.release()
+
+    def _execute_locked(
+        self,
+        store: DatasetStore,
+        task_executor: TaskExecutor,
+        last_modified_at_map: Optional[DatasetLastModifiedAtMap] = None,
     ) -> Iterator[IngestionJobSummary]:
         is_first_chunk = True
         ingestion_job_summary = IngestionJobSummary.new(ingestion_job=self)
@@ -498,7 +527,14 @@ def execute(
                             "StopProcessing raised — saving partial results "
                             "and stopping"
                         )
-                        ingestion_job_summary.set_finished()
+                        ingestion_job_summary.set_aborted()
+                        yield ingestion_job_summary
+                        raise
+                    except (KeyboardInterrupt, SystemExit):
+                        logger.warning(
+                            "Interrupted — saving partial results and aborting"
+                        )
+                        ingestion_job_summary.set_aborted()
                         yield ingestion_job_summary
                         raise
                     except FatalError as e:
@@ -636,7 +672,12 @@ def filtered_stream():
                     self._save_progress(store, ingestion_job_summary)
         except StopProcessing:
             logger.info("StopProcessing raised — saving partial results and stopping")
-            ingestion_job_summary.set_finished()
+            ingestion_job_summary.set_aborted()
+            yield ingestion_job_summary
+            raise
+        except (KeyboardInterrupt, SystemExit):
+            logger.warning("Interrupted — saving partial results and aborting")
+            ingestion_job_summary.set_aborted()
             yield ingestion_job_summary
             raise
         except FatalError as e:
diff --git a/ingestify/domain/models/ingestion/ingestion_job_summary.py b/ingestify/domain/models/ingestion/ingestion_job_summary.py
index c3446be..9a10ddb 100644
--- a/ingestify/domain/models/ingestion/ingestion_job_summary.py
+++ b/ingestify/domain/models/ingestion/ingestion_job_summary.py
@@ -20,6 +20,10 @@ class IngestionJobState(str, Enum):
     FINISHED = "FINISHED"
     SKIPPED = "SKIPPED"
     FAILED = "FAILED"
+    # Stopped before the whole plan finished -- a source raising StopProcessing (e.g.
+    # quota), Ctrl-C, or a Cloud Run task timeout (SIGTERM). Partial results are saved and
+    # the next run resumes; it is NOT an error (that is FAILED).
+    ABORTED = "ABORTED"
 
 
 def format_duration(duration: timedelta):
@@ -119,6 +123,10 @@ def set_skipped(self):
         self.state = IngestionJobState.SKIPPED
         self._set_ended()
 
+    def set_aborted(self):
+        self.state = IngestionJobState.ABORTED
+        self._set_ended()
+
     @property
     def duration(self) -> timedelta:
         # ended_at is None while the job is still RUNNING (live snapshots);
diff --git a/ingestify/infra/store/dataset/sqlalchemy/repository.py b/ingestify/infra/store/dataset/sqlalchemy/repository.py
index 00775fb..2a5cd1b 100644
--- a/ingestify/infra/store/dataset/sqlalchemy/repository.py
+++ b/ingestify/infra/store/dataset/sqlalchemy/repository.py
@@ -1,3 +1,4 @@
+import hashlib
 import itertools
 import json
 import logging
@@ -26,6 +27,7 @@
 from sqlalchemy.orm import Session, Query, sessionmaker, scoped_session
 
 from ingestify.domain import File, Revision
+from ingestify.domain.models.dataset.dataset_repository import RunLock
 from ingestify.domain.models.dataset.revision import RevisionState
 from ingestify.domain.models import (
     Dataset,
@@ -205,6 +207,34 @@ def get(self):
         return self.session()
 
 
+def _advisory_key(job_key: str) -> int:
+    """Stable signed 64-bit key for Postgres pg_advisory_lock (which takes a bigint)."""
+    return int.from_bytes(
+        hashlib.blake2b(job_key.encode(), digest_size=8).digest(), "big", signed=True
+    )
+
+
+def _lock_name(job_key: str) -> str:
+    """Stable MySQL GET_LOCK name (user-level lock names must be <= 64 chars)."""
+    return "ingestify_" + hashlib.blake2b(job_key.encode(), digest_size=24).hexdigest()
+
+
+class _SessionRunLock(RunLock):
+    """Holds a dedicated connection whose session owns the lock. ``release()`` unlocks and
+    returns the connection; a dropped connection (crash / process end) frees it too."""
+
+    def __init__(self, conn: Connection, unlock_sql, params: dict):
+        self._conn = conn
+        self._unlock_sql = unlock_sql
+        self._params = params
+
+    def release(self) -> None:
+        try:
+            self._conn.execute(self._unlock_sql, self._params)
+        finally:
+            self._conn.close()
+
+
 class SqlAlchemyDatasetRepository(DatasetRepository):
     def __init__(
         self, session_provider: SqlAlchemySessionProvider, identifier_transformer=None
@@ -223,6 +253,43 @@ def session(self):
     def dialect(self) -> Dialect:
         return self.session_provider.dialect
 
+    def acquire_run_lock(self, job_key: str) -> Optional[RunLock]:
+        dialect = self.dialect.name
+        if dialect not in ("postgresql", "mysql"):
+            # No cross-process lock (e.g. SQLite): fall back to the no-op.
+            return super().acquire_run_lock(job_key)
+
+        # A dedicated AUTOCOMMIT connection: the session-level lock is held for the life of
+        # this connection (released the instant it/the process ends), with no long-open
+        # transaction. Kept outside the scoped_session, which is reused/closed.
+        conn = self.session_provider.engine.connect().execution_options(
+            isolation_level="AUTOCOMMIT"
+        )
+        try:
+            if dialect == "postgresql":
+                key = _advisory_key(job_key)
+                acquired = bool(
+                    conn.execute(
+                        text("SELECT pg_try_advisory_lock(:k)"), {"k": key}
+                    ).scalar()
+                )
+                unlock, params = text("SELECT pg_advisory_unlock(:k)"), {"k": key}
+            else:  # mysql
+                name = _lock_name(job_key)
+                acquired = (
+                    conn.execute(text("SELECT GET_LOCK(:n, 0)"), {"n": name}).scalar()
+                    == 1
+                )
+                unlock, params = text("SELECT RELEASE_LOCK(:n)"), {"n": name}
+        except Exception:
+            conn.close()
+            raise
+
+        if not acquired:
+            conn.close()
+            return None
+        return _SessionRunLock(conn, unlock, params)
+
     @property
     def dataset_table(self):
         return self.session_provider.dataset_table
diff --git a/ingestify/tests/test_fatal_error.py b/ingestify/tests/test_fatal_error.py
index b95e615..a2b150a 100644
--- a/ingestify/tests/test_fatal_error.py
+++ b/ingestify/tests/test_fatal_error.py
@@ -167,8 +167,8 @@ def test_fatal_error_in_async_collect_persists_failed_summary(tmp_path):
 
 
 def test_stop_processing_in_async_collect_persists_summary(tmp_path):
-    """StopProcessing while collecting is a controlled stop: the summary is
-    persisted (FINISHED) instead of being lost, mirroring the sync path."""
+    """StopProcessing while collecting is a controlled early stop: the summary is
+    persisted as ABORTED (not lost, not FINISHED), mirroring the sync path."""
     engine = _run_async(
         _AsyncSource("s", ["a", "b"], StopProcessing("quota")), tmp_path
     )
@@ -178,4 +178,4 @@ def test_stop_processing_in_async_collect_persists_summary(tmp_path):
 
     summaries = engine.store.dataset_repository.load_ingestion_job_summaries()
     assert len(summaries) == 1
-    assert summaries[0].state == IngestionJobState.FINISHED
+    assert summaries[0].state == IngestionJobState.ABORTED
diff --git a/ingestify/tests/test_run_lock.py b/ingestify/tests/test_run_lock.py
new file mode 100644
index 0000000..bcfe87b
--- /dev/null
+++ b/ingestify/tests/test_run_lock.py
@@ -0,0 +1,160 @@
+"""Single-run lock — one (source, dataset_type, selector) job must never run in two
+processes at once (design: docs/design/single-run-lock.md).
+
+WHERE the lock lives, and WHY there:
+  - In the metadata store (`DatasetRepository.acquire_run_lock`) — the only shared,
+    cross-process state ingestify has. On a server DB it is a session-scoped lock held on a
+    dedicated connection (Postgres `pg_advisory_lock`, MySQL `GET_LOCK`), so it is released
+    the instant that connection (i.e. the process) ends: no TTL, no reaper. On stores without
+    one (SQLite / local single-process) it is a documented no-op.
+  - Taken in `IngestionJob.execute`, before the RUNNING summary is written, so a run that
+    loses the race posts nothing and never shows up as RUNNING.
+
+Session locks are a server-DB feature, so the exclusivity test runs against Postgres/MySQL
+and skips on SQLite; the no-op contract is checked where no such lock exists.
+"""
+import pytest
+
+from ingestify import DatasetResource, Source
+from ingestify.domain import DataSpecVersionCollection
+from ingestify.domain.models import Selector
+from ingestify.domain.models.fetch_policy import FetchPolicy
+from ingestify.domain.models.ingestion.ingestion_job_summary import IngestionJobState
+from ingestify.domain.models.ingestion.ingestion_plan import IngestionPlan
+from ingestify.main import get_engine
+from ingestify.utils import utcnow
+
+_LOCKING_DIALECTS = ("postgresql", "mysql")
+
+
+def test_run_lock_is_exclusive_and_releases(engine):
+    repo = engine.store.dataset_repository
+    if repo.dialect.name not in _LOCKING_DIALECTS:
+        pytest.skip("session-lock mutex needs Postgres or MySQL")
+
+    first = repo.acquire_run_lock("source_a:type_x:{}")
+    assert first is not None  # free -> acquired
+    assert repo.acquire_run_lock("source_a:type_x:{}") is None  # 2nd session -> blocked
+
+    first.release()
+    again = repo.acquire_run_lock("source_a:type_x:{}")  # released -> free again
+    assert again is not None
+
+    other = repo.acquire_run_lock("source_b:type_y:{}")  # different job -> independent
+    assert other is not None
+    again.release()
+    other.release()
+
+
+def test_run_lock_is_noop_without_cross_process_store(engine):
+    repo = engine.store.dataset_repository
+    if repo.dialect.name in _LOCKING_DIALECTS:
+        pytest.skip("covered by the exclusivity test")
+
+    # No cross-process lock (e.g. SQLite): acquire must always grant, so a single local
+    # process is never blocked by itself.
+    a = repo.acquire_run_lock("x:y:{}")
+    b = repo.acquire_run_lock("x:y:{}")
+    assert a is not None and b is not None
+    a.release()
+    b.release()
+
+
+class _FakeSource(Source):
+    """Ingests one dataset; ``find_datasets_called`` records whether its job actually ran."""
+
+    provider = "run_lock_fake"
+
+    def __init__(self, name):
+        super().__init__(name)
+        self.find_datasets_called = False
+
+    def find_datasets(
+        self, dataset_type, data_spec_versions, dataset_collection_metadata, **kwargs
+    ):
+        self.find_datasets_called = True
+        yield DatasetResource(
+            dataset_resource_id={"keyword": "k"},
+            provider=self.provider,
+            dataset_type="keyword",
+            name="k",
+        ).add_file(
+            last_modified=utcnow(),
+            data_feed_key="data",
+            data_spec_version="v1",
+            json_content={"keyword": "k"},
+        )
+
+
+def _engine(db_url, file_dir, source):
+    engine = get_engine(
+        metadata_url=db_url,
+        file_url=f"file://{file_dir}",
+        bucket="main",
+        disable_events=True,
+    )
+    dsv = DataSpecVersionCollection.from_dict({"default": "v1"})
+    engine.add_ingestion_plan(
+        IngestionPlan(
+            source=source,
+            dataset_type="keyword",
+            selectors=[Selector.build({}, data_spec_versions=dsv)],
+            fetch_policy=FetchPolicy(),
+            data_spec_versions=dsv,
+        )
+    )
+    return engine
+
+
+def _summaries(engine):
+    return engine.store.dataset_repository.load_ingestion_job_summaries()
+
+
+def _cleanup(*engines):
+    providers = [e.store.dataset_repository.session_provider for e in engines]
+    # Close every session + pool first (releasing any lingering transaction/lock), then
+    # DROP on a fresh connection -- otherwise DROP TABLE would wait on an idle connection.
+    for sp in providers:
+        sp.session.remove()
+        sp.engine.dispose()
+    providers[0].drop_all_tables()
+
+
+def test_two_instances_same_job_only_one_runs(ingestify_test_database_url, tmp_path):
+    """Two ingestify instances over the SAME database contend for the SAME
+    (source, dataset_type, selector) job. Instance 1 holds the run lock (on its own
+    connection, exactly as a running instance would); instance 2 tries the job and must
+    skip -- its source is never invoked and a SKIPPED summary is recorded. Once the lock is
+    freed, a fresh instance runs the job normally. Needs a cross-process lock, so it runs on
+    Postgres/MySQL and skips on SQLite. No threads -> no cleanup deadlock."""
+    if not ingestify_test_database_url.startswith(("postgres", "mysql")):
+        pytest.skip("cross-process lock needs Postgres or MySQL")
+
+    dsv = DataSpecVersionCollection.from_dict({"default": "v1"})
+    # Mirrors IngestionJob._job_key for this plan (source, dataset_type, empty selector).
+    job_key = f"shared:keyword:{Selector.build({}, data_spec_versions=dsv).key}"
+
+    holder = _engine(ingestify_test_database_url, tmp_path, _FakeSource("shared"))
+    blocked_source = _FakeSource("shared")
+    blocked = _engine(ingestify_test_database_url, tmp_path, blocked_source)
+    free_source = _FakeSource("shared")
+    free = _engine(ingestify_test_database_url, tmp_path, free_source)
+
+    try:
+        lock = holder.store.acquire_run_lock(job_key)  # instance 1 holds the lock
+        assert lock is not None
+        try:
+            blocked.run()  # instance 2, same job, lock held -> must skip
+            assert blocked_source.find_datasets_called is False  # it never ran the job
+            states = [s.state for s in _summaries(blocked)]
+            assert states.count(IngestionJobState.SKIPPED) == 1
+            assert IngestionJobState.FINISHED not in states
+        finally:
+            lock.release()  # instance 1 done -> lock free
+
+        free.run()  # lock free now -> the job runs normally
+        assert free_source.find_datasets_called is True
+        finished = [s.state for s in _summaries(free)].count(IngestionJobState.FINISHED)
+        assert finished == 1
+    finally:
+        _cleanup(holder, blocked, free)
diff --git a/ingestify/tests/test_stop_processing.py b/ingestify/tests/test_stop_processing.py
index 8837e05..3a6391f 100644
--- a/ingestify/tests/test_stop_processing.py
+++ b/ingestify/tests/test_stop_processing.py
@@ -9,6 +9,7 @@
     DatasetCollectionMetadata,
 )
 from ingestify.domain.models.fetch_policy import FetchPolicy
+from ingestify.domain.models.ingestion.ingestion_job_summary import IngestionJobState
 from ingestify.domain.models.ingestion.ingestion_plan import IngestionPlan
 from ingestify.exceptions import StopProcessing
 from ingestify.utils import utcnow
@@ -22,16 +23,21 @@ def stopping_loader(file_resource, current_file, **kwargs):
     raise StopProcessing("API quota exhausted")
 
 
+def interrupting_loader(file_resource, current_file, **kwargs):
+    raise KeyboardInterrupt()
+
+
 class SourceWithStopProcessing(Source):
     """Source that yields 5 datasets. The 3rd one raises StopProcessing."""
 
     provider = "test_provider"
+    bad_loader = staticmethod(stopping_loader)
 
     def find_datasets(
         self, dataset_type, data_spec_versions, dataset_collection_metadata, **kwargs
     ):
         for i in range(5):
-            loader = stopping_loader if i == 2 else good_loader
+            loader = self.bad_loader if i == 2 else good_loader
             r = DatasetResource(
                 dataset_resource_id={"item_id": i},
                 provider=self.provider,
@@ -103,3 +109,35 @@ def test_stop_processing_saves_ingestion_job_summary(engine):
         assert (
             mock_save.call_count >= 1
         ), "save_ingestion_job_summary should be called even on StopProcessing"
+
+
+class SourceWithInterrupt(SourceWithStopProcessing):
+    """Same shape, but the 3rd dataset raises KeyboardInterrupt (Ctrl-C / SIGTERM)."""
+
+    bad_loader = staticmethod(interrupting_loader)
+
+
+def _states(engine):
+    return [
+        s.state for s in engine.store.dataset_repository.load_ingestion_job_summaries()
+    ]
+
+
+def test_stop_processing_marks_summary_aborted(engine):
+    """A StopProcessing stop leaves the summary ABORTED (stopped early), not FINISHED."""
+    _setup(engine, SourceWithStopProcessing("s"))
+
+    with pytest.raises(StopProcessing):
+        engine.run()
+
+    assert _states(engine) == [IngestionJobState.ABORTED]
+
+
+def test_keyboard_interrupt_marks_summary_aborted(engine):
+    """Ctrl-C mid-run leaves the summary ABORTED instead of a stuck RUNNING row."""
+    _setup(engine, SourceWithInterrupt("s"))
+
+    with pytest.raises(KeyboardInterrupt):
+        engine.run()
+
+    assert _states(engine) == [IngestionJobState.ABORTED]