From ac35d8df8fa46976cd45448db7d5c9cdcb10bd0b Mon Sep 17 00:00:00 2001
From: "jeremy.barisch.rooney@channable.com"
 <jeremy.barisch.rooney@channable.com>
Date: Mon, 8 Jun 2026 13:57:48 +0200
Subject: [PATCH 01/16] Add
 ProducerClient.lookup_submission_ids_by_strategic_metadata

---
 .../python/opsqueue/producer.py               | 23 ++++++++
 libs/opsqueue_python/src/producer.rs          | 18 +++++++
 libs/opsqueue_python/tests/test_roundtrip.py  | 52 +++++++++++++++++++
 opsqueue/src/common/submission.rs             | 32 +++++++++++-
 opsqueue/src/producer/client.rs               | 28 ++++++++++
 opsqueue/src/producer/server.rs               | 16 ++++++
 6 files changed, 167 insertions(+), 2 deletions(-)

diff --git a/libs/opsqueue_python/python/opsqueue/producer.py b/libs/opsqueue_python/python/opsqueue/producer.py
index 5c6a7f06..307a2643 100644
--- a/libs/opsqueue_python/python/opsqueue/producer.py
+++ b/libs/opsqueue_python/python/opsqueue/producer.py
@@ -42,6 +42,10 @@
 ]
 
 
+class LookupIdsWithEmptyStrategicMetadataError(Exception):
+    pass
+
+
 class ProducerClient:
     """
     Opsqueue producer client. Allows sending of large collections of operations ('submissions')
@@ -367,6 +371,25 @@ def lookup_submission_id_by_prefix(self, prefix: str) -> SubmissionId | None:
         """
         return self.inner.lookup_submission_id_by_prefix(prefix)
 
+    def lookup_submission_ids_by_strategic_metadata(
+        self, strategic_metadata: dict[str, int]
+    ) -> list[SubmissionId]:
+        """Attempts to find in-progress submissions where the strategic metadata
+        of that submission includes all of the key-value pairs of the given
+        'strategic_metadata'. A matching submission must include all of the
+        given key-value pairs, but it may also contain other key-value pairs.
+
+        Raises:
+        - `LookupIdsWithEmptyStrategicMetadataError` if the provided
+          'strategic_metadata' contained no key-value pairs to look for.
+
+        """
+        if len(strategic_metadata) == 0:
+            raise LookupIdsWithEmptyStrategicMetadataError()
+        return self.inner.lookup_submission_ids_by_strategic_metadata(  # type: ignore[no-any-return]
+            strategic_metadata
+        )
+
     def is_completed(self, submission_id: SubmissionId) -> bool:
         raise NotImplementedError
 
diff --git a/libs/opsqueue_python/src/producer.rs b/libs/opsqueue_python/src/producer.rs
index 5598172a..212e9438 100644
--- a/libs/opsqueue_python/src/producer.rs
+++ b/libs/opsqueue_python/src/producer.rs
@@ -190,6 +190,24 @@ impl ProducerClient {
         })
     }
 
+    /// Attempts to find the IDs of submission matching ALL key-values pairs of
+    /// the given strategic metadata.
+    pub fn lookup_submission_ids_by_strategic_metadata(
+        &self,
+        py: Python<'_>,
+        strategic_metadata: StrategicMetadataMap,
+    ) -> CPyResult<Vec<SubmissionId>, E<FatalPythonException, InternalProducerClientError>> {
+        py.allow_threads(|| {
+            self.block_unless_interrupted(async {
+                self.producer_client
+                    .lookup_submission_ids_by_strategic_metadata(&strategic_metadata)
+                    .await
+                    .map(|res| res.into_iter().map(Into::into).collect())
+                    .map_err(|e| CError(R(e)))
+            })
+        })
+    }
+
     /// Directly inserts a submission without sending the chunks to GCS
     /// (but immediately embedding them in the DB).
     /// NOTE: This does not support StrategicMetadata currently
diff --git a/libs/opsqueue_python/tests/test_roundtrip.py b/libs/opsqueue_python/tests/test_roundtrip.py
index ad105414..ddb97d3a 100644
--- a/libs/opsqueue_python/tests/test_roundtrip.py
+++ b/libs/opsqueue_python/tests/test_roundtrip.py
@@ -4,6 +4,7 @@
 
 from collections.abc import Iterator, Sequence
 from opsqueue.producer import (
+    LookupIdsWithEmptyStrategicMetadataError,
     SubmissionId,
     ProducerClient,
     SubmissionCompleted,
@@ -508,3 +509,54 @@ def consume(x: int) -> int | None:
         with pytest.raises(SubmissionFailedError) as exc_info:
             producer_client.blocking_stream_completed_submission(submission_id)
         assert exc_info.value.submission.chunks_done == len(chunks) - 1
+
+
+def test_lookup_submission_ids_by_strategic_metadata(opsqueue: OpsqueueProcess) -> None:
+    """Lookup of submission IDs should only match in progress submissions with
+    all pieces of strategic metadata.
+
+    """
+    url = "file:///tmp/opsqueue/test_lookup_submission_ids_by_strategic_metadata"
+    producer_client = ProducerClient(f"localhost:{opsqueue.port}", url)
+    id_1 = producer_client.insert_submission(
+        [1], chunk_size=1, strategic_metadata={"foo": 1, "bar": 2, "wow": 3}
+    )
+    id_2 = producer_client.insert_submission(
+        [1], chunk_size=1, strategic_metadata={"foo": 1, "bar": 2, "moo": 3}
+    )
+    # Inserting some similar data to that above, which shouldn't get matched.
+    producer_client.insert_submission(
+        [1], chunk_size=1, strategic_metadata={"foo": 2, "bar": 1}
+    )
+
+    def test_lookup(
+        strategic_metadata: dict[str, int], expected_ids: list[int]
+    ) -> None:
+        found_ids = producer_client.lookup_submission_ids_by_strategic_metadata(
+            strategic_metadata
+        )
+        assert isinstance(found_ids, list)
+        assert all(map(lambda x: isinstance(x, SubmissionId), found_ids))
+        assert found_ids == expected_ids
+
+    test_lookup({"foo": 1}, [id_1, id_2])
+    test_lookup({"foo": 1, "bar": 2}, [id_1, id_2])
+    test_lookup({"foo": 1, "MISS": 2}, [])
+    test_lookup({"wow": 3}, [id_1])
+
+    # Should only match in-progress submission.
+    producer_client.cancel_submission(id_1)
+    test_lookup({"foo": 1}, [id_2])
+
+
+def test_lookup_submission_ids_by_empty_strategic_metadata(
+    opsqueue: OpsqueueProcess,
+) -> None:
+    """Lookup of submission IDs with empty strategic_metadata should throw a
+    LookupIdsWithEmptyStrategicMetadataError.
+
+    """
+    url = "file:///tmp/opsqueue/test_lookup_submission_ids_by_empty_strategic_metadata"
+    producer_client = ProducerClient(f"localhost:{opsqueue.port}", url)
+    with pytest.raises(LookupIdsWithEmptyStrategicMetadataError):
+        producer_client.lookup_submission_ids_by_strategic_metadata({})
diff --git a/opsqueue/src/common/submission.rs b/opsqueue/src/common/submission.rs
index 5355f646..d2454156 100644
--- a/opsqueue/src/common/submission.rs
+++ b/opsqueue/src/common/submission.rs
@@ -275,8 +275,7 @@ pub mod db {
         db::{Connection, True, WriterConnection, WriterPool},
     };
     use chunk::ChunkSize;
-    use sqlx::{query, Sqlite};
-
+    use sqlx::{query, query_as, QueryBuilder, Row, Sqlite};
     use axum_prometheus::metrics::{counter, histogram};
 
     use super::*;
@@ -544,6 +543,35 @@ pub mod db {
         Ok(row.map(|row| row.id))
     }
 
+    pub async fn lookup_ids_by_strategic_metadata(
+        strategic_metadata: StrategicMetadataMap,
+        mut conn: impl Connection,
+    ) -> Result<Vec<SubmissionId>, DatabaseError> {
+        // SQLx currently only supports "WHERE X IN (a, ...)" queries for postgres:
+        // https://github.com/transact-rs/sqlx/blob/main/FAQ.md#how-can-i-do-a-select--where-foo-in--query
+        // So we workaround this by manually building the query, foregoing
+        // sqlx's nice type-checking.
+        let mut query_builder: QueryBuilder<Sqlite> = QueryBuilder::new(
+            "
+            SELECT submission_id
+            FROM submissions_metadata
+            INNER JOIN submissions on submissions.id = submission_id
+            WHERE (metadata_key, metadata_value) IN (
+        ",
+        );
+        query_builder.push_values(strategic_metadata.iter(), |mut b, sm| {
+            b.push_bind(sm.0).push_bind(sm.1);
+        });
+        query_builder.push(") GROUP BY submission_id HAVING count(*) = ");
+        query_builder.push_bind(strategic_metadata.len() as i64);
+        query_builder.push(" ORDER BY submission_id");
+        let rows = query_builder.build().fetch_all(conn.get_inner()).await?;
+        Ok(rows
+            .into_iter()
+            .map(|row| row.get("submission_id"))
+            .collect())
+    }
+
     #[tracing::instrument(skip(conn))]
     pub async fn submission_status(
         id: SubmissionId,
diff --git a/opsqueue/src/producer/client.rs b/opsqueue/src/producer/client.rs
index b09e20b0..2bfa9052 100644
--- a/opsqueue/src/producer/client.rs
+++ b/opsqueue/src/producer/client.rs
@@ -10,6 +10,7 @@ use crate::{
         errors::E::{L, R},
         errors::{SubmissionNotCancellable, SubmissionNotFound},
         submission::{SubmissionId, SubmissionStatus},
+        StrategicMetadataMap,
     },
     tracing::CarrierMap,
     E,
@@ -226,6 +227,33 @@ impl Client {
         .await
     }
 
+    pub async fn lookup_submission_ids_by_strategic_metadata(
+        &self,
+        strategic_metadata: &StrategicMetadataMap,
+    ) -> Result<Vec<SubmissionId>, InternalProducerClientError> {
+        (|| async {
+            let base_url = &self.base_url;
+            let resp = self
+                .http_client
+                .post(format!(
+                    "{base_url}/submissions/lookup_ids_by_strategic_metadata"
+                ))
+                .json(strategic_metadata)
+                .send()
+                .await?
+                .error_for_status()?;
+            let bytes = resp.bytes().await?;
+            let body = serde_json::from_slice(&bytes)?;
+            Ok(body)
+        })
+        .retry(retry_policy())
+        .when(InternalProducerClientError::is_ephemeral)
+        .notify(|err, dur| {
+            tracing::debug!("retrying error {err:?} with sleeping {dur:?}");
+        })
+        .await
+    }
+
     /// Get the server's version from the `/version` endpoint.
     ///
     /// A successful result will be the value of [`VERSION_CARGO_SEMVER`][crate::VERSION_CARGO_SEMVER]
diff --git a/opsqueue/src/producer/server.rs b/opsqueue/src/producer/server.rs
index 0298a2b0..49865055 100644
--- a/opsqueue/src/producer/server.rs
+++ b/opsqueue/src/producer/server.rs
@@ -2,7 +2,9 @@ use std::sync::Arc;
 
 use crate::common::errors::E::{L, R};
 use crate::common::submission::{self, SubmissionId};
+use crate::common::StrategicMetadataMap;
 use crate::db::{self, DBPools};
+use axum::extract;
 use axum::extract::{Path, State};
 use axum::http::StatusCode;
 use axum::response::{IntoResponse, Response};
@@ -60,6 +62,10 @@ impl ServerState {
                 "/submissions/lookup_id_by_prefix/{prefix}",
                 get(lookup_submission_id_by_prefix),
             )
+            .route(
+                "/submissions/lookup_ids_by_strategic_metadata",
+                post(lookup_submission_ids_by_strategic_metadata),
+            )
             .route("/submissions/{submission_id}", get(submission_status))
             .route("/version", get(crate::server::version_endpoint)) // We're also exposing it here so the producer client can view it
             .with_state(self)
@@ -133,6 +139,16 @@ async fn lookup_submission_id_by_prefix(
     Ok(Json(submission_id))
 }
 
+async fn lookup_submission_ids_by_strategic_metadata(
+    State(state): State<ServerState>,
+    extract::Json(strategic_metadata): extract::Json<StrategicMetadataMap>,
+) -> Result<Json<Vec<SubmissionId>>, ServerError> {
+    let mut conn = state.pool.reader_conn().await?;
+    let submission_ids =
+        submission::db::lookup_ids_by_strategic_metadata(strategic_metadata, &mut conn).await?;
+    Ok(Json(submission_ids))
+}
+
 #[tracing::instrument(level = "debug", skip(state))]
 async fn insert_submission(
     State(state): State<ServerState>,

From 3f5b86f0aa1501d84fd2e0dbd202539b590ed890 Mon Sep 17 00:00:00 2001
From: "jeremy.barisch.rooney@channable.com"
 <jeremy.barisch.rooney@channable.com>
Date: Tue, 7 Jul 2026 13:55:45 +0200
Subject: [PATCH 02/16] fixup! Add
 ProducerClient.lookup_submission_ids_by_strategic_metadata

Document InternalProducerClientError raised
---
 libs/opsqueue_python/python/opsqueue/producer.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/libs/opsqueue_python/python/opsqueue/producer.py b/libs/opsqueue_python/python/opsqueue/producer.py
index 307a2643..340c1afd 100644
--- a/libs/opsqueue_python/python/opsqueue/producer.py
+++ b/libs/opsqueue_python/python/opsqueue/producer.py
@@ -382,6 +382,7 @@ def lookup_submission_ids_by_strategic_metadata(
         Raises:
         - `LookupIdsWithEmptyStrategicMetadataError` if the provided
           'strategic_metadata' contained no key-value pairs to look for.
+        - `InternalProducerClientError` if there is a low-level internal error.
 
         """
         if len(strategic_metadata) == 0:

From c124c4b76c5157825a1b1f595c7cc16347894ac8 Mon Sep 17 00:00:00 2001
From: "jeremy.barisch.rooney@channable.com"
 <jeremy.barisch.rooney@channable.com>
Date: Tue, 7 Jul 2026 14:42:57 +0200
Subject: [PATCH 03/16] fixup! Add
 ProducerClient.lookup_submission_ids_by_strategic_metadata

Support empty strategic metadata
---
 libs/opsqueue_python/python/opsqueue/producer.py |  4 ----
 libs/opsqueue_python/tests/test_roundtrip.py     | 10 ++++++----
 opsqueue/src/common/submission.rs                | 13 ++++++++++++-
 3 files changed, 18 insertions(+), 9 deletions(-)

diff --git a/libs/opsqueue_python/python/opsqueue/producer.py b/libs/opsqueue_python/python/opsqueue/producer.py
index 340c1afd..11adfe75 100644
--- a/libs/opsqueue_python/python/opsqueue/producer.py
+++ b/libs/opsqueue_python/python/opsqueue/producer.py
@@ -380,13 +380,9 @@ def lookup_submission_ids_by_strategic_metadata(
         given key-value pairs, but it may also contain other key-value pairs.
 
         Raises:
-        - `LookupIdsWithEmptyStrategicMetadataError` if the provided
-          'strategic_metadata' contained no key-value pairs to look for.
         - `InternalProducerClientError` if there is a low-level internal error.
 
         """
-        if len(strategic_metadata) == 0:
-            raise LookupIdsWithEmptyStrategicMetadataError()
         return self.inner.lookup_submission_ids_by_strategic_metadata(  # type: ignore[no-any-return]
             strategic_metadata
         )
diff --git a/libs/opsqueue_python/tests/test_roundtrip.py b/libs/opsqueue_python/tests/test_roundtrip.py
index ddb97d3a..9ea4f2fb 100644
--- a/libs/opsqueue_python/tests/test_roundtrip.py
+++ b/libs/opsqueue_python/tests/test_roundtrip.py
@@ -552,11 +552,13 @@ def test_lookup(
 def test_lookup_submission_ids_by_empty_strategic_metadata(
     opsqueue: OpsqueueProcess,
 ) -> None:
-    """Lookup of submission IDs with empty strategic_metadata should throw a
-    LookupIdsWithEmptyStrategicMetadataError.
+    """Lookup of submission IDs with empty strategic_metadata should NOT raise
+    an exception.
 
     """
     url = "file:///tmp/opsqueue/test_lookup_submission_ids_by_empty_strategic_metadata"
     producer_client = ProducerClient(f"localhost:{opsqueue.port}", url)
-    with pytest.raises(LookupIdsWithEmptyStrategicMetadataError):
-        producer_client.lookup_submission_ids_by_strategic_metadata({})
+    count = 6
+    for _ in range(count):
+        producer_client.insert_submission([1], chunk_size=1)
+    assert len(producer_client.lookup_submission_ids_by_strategic_metadata({})) == count
diff --git a/opsqueue/src/common/submission.rs b/opsqueue/src/common/submission.rs
index d2454156..3f52ac14 100644
--- a/opsqueue/src/common/submission.rs
+++ b/opsqueue/src/common/submission.rs
@@ -275,7 +275,7 @@ pub mod db {
         db::{Connection, True, WriterConnection, WriterPool},
     };
     use chunk::ChunkSize;
-    use sqlx::{query, query_as, QueryBuilder, Row, Sqlite};
+    use sqlx::{QueryBuilder, Row, Sqlite, query, query_scalar};
     use axum_prometheus::metrics::{counter, histogram};
 
     use super::*;
@@ -547,6 +547,17 @@ pub mod db {
         strategic_metadata: StrategicMetadataMap,
         mut conn: impl Connection,
     ) -> Result<Vec<SubmissionId>, DatabaseError> {
+        // The main query below to match on strategic_metadata will fail at
+        // run-time if strategic_metadata is empty (because the query will
+        // produce `IN ()`). So we handle the empty case here separately.
+        if strategic_metadata.is_empty() {
+            let ids = query_scalar!(
+                r#"SELECT id AS "id: SubmissionId" FROM submissions ORDER by id"#,
+            )
+            .fetch_all(conn.get_inner())
+            .await?;
+            return Ok(ids);
+        }
         // SQLx currently only supports "WHERE X IN (a, ...)" queries for postgres:
         // https://github.com/transact-rs/sqlx/blob/main/FAQ.md#how-can-i-do-a-select--where-foo-in--query
         // So we workaround this by manually building the query, foregoing

From fca2f407ef23edeb8ba5520448e457f9c6b1cde8 Mon Sep 17 00:00:00 2001
From: "jeremy.barisch.rooney@channable.com"
 <jeremy.barisch.rooney@channable.com>
Date: Tue, 7 Jul 2026 17:47:56 +0200
Subject: [PATCH 04/16] fixup! Add
 ProducerClient.lookup_submission_ids_by_strategic_metadata

Add configurable limit to submissions lookup
---
 justfile                                      |  4 +-
 .../python/opsqueue/exceptions.py             | 28 +++++++++++
 .../python/opsqueue/producer.py               |  5 ++
 libs/opsqueue_python/src/errors.rs            |  9 +++-
 libs/opsqueue_python/src/producer.rs          | 11 ++++-
 libs/opsqueue_python/tests/test_roundtrip.py  |  1 -
 opsqueue/src/common/errors.rs                 |  4 ++
 opsqueue/src/common/mod.rs                    | 47 +++++++++++++++++++
 opsqueue/src/common/submission.rs             | 47 ++++++++++++++-----
 opsqueue/src/config.rs                        | 13 +++++
 opsqueue/src/producer/client.rs               | 39 +++++++++++----
 opsqueue/src/producer/server.rs               | 37 +++++++++++----
 opsqueue/src/server.rs                        |  8 +++-
 13 files changed, 216 insertions(+), 37 deletions(-)

diff --git a/justfile b/justfile
index cdba3709..65243ab6 100644
--- a/justfile
+++ b/justfile
@@ -47,7 +47,7 @@ test-integration *TEST_ARGS: build-bin build-python
   cd libs/opsqueue_python
   source "./.setup_local_venv.sh"
 
-  timeout 600 pytest --color=yes {{TEST_ARGS}}
+  timeout 60 pytest --color=yes {{TEST_ARGS}}
 
 # Python integration test suite, using artefacts built through Nix. Args are forwarded to pytest
 [group('nix')]
@@ -61,7 +61,7 @@ nix-test-integration *TEST_ARGS: nix-build-bin
   export OPSQUEUE_VIA_NIX=true
   export RUST_LOG="opsqueue=debug"
 
-  timeout 600 pytest --color=yes {{TEST_ARGS}}
+  timeout 60 pytest --color=yes {{TEST_ARGS}}
 
 # Run all linters, fast and slow
 [group('lint')]
diff --git a/libs/opsqueue_python/python/opsqueue/exceptions.py b/libs/opsqueue_python/python/opsqueue/exceptions.py
index e1ae2bb3..615dc958 100644
--- a/libs/opsqueue_python/python/opsqueue/exceptions.py
+++ b/libs/opsqueue_python/python/opsqueue/exceptions.py
@@ -153,6 +153,34 @@ class SubmissionNotCompletedYetError(IncorrectUsageError):
     pass
 
 
+class TooManyMatchingSubmissionsError(IncorrectUsageError):
+    """
+    Raised when a strategic-metadata lookup matches more submissions
+    than the server's configured maximum (``max_submissions_returned``).
+
+    Narrow the query with more specific strategic metadata, or raise the
+    server's configured maximum.
+    """
+
+    __slots__ = ["max_submissions"]
+
+    def __init__(
+        self,
+        max_submissions: int,
+    ):
+        super().__init__()
+        self.max_submissions = max_submissions
+
+    def __str__(self) -> str:
+        return (
+            f"The lookup matched more submissions than the configured "
+            f"maximum of {self.max_submissions}"
+        )
+
+    def __repr__(self) -> str:
+        return str(self)
+
+
 # Internal errors:
 
 
diff --git a/libs/opsqueue_python/python/opsqueue/producer.py b/libs/opsqueue_python/python/opsqueue/producer.py
index 11adfe75..7a93da5e 100644
--- a/libs/opsqueue_python/python/opsqueue/producer.py
+++ b/libs/opsqueue_python/python/opsqueue/producer.py
@@ -18,6 +18,7 @@
     SubmissionFailedError,
     SubmissionNotCancellableError,
     SubmissionNotFoundError,
+    TooManyMatchingSubmissionsError,
 )
 from .opsqueue_internal import (  # type: ignore[import-not-found]
     SubmissionId,
@@ -38,6 +39,7 @@
     "SubmissionNotCancellable",
     "SubmissionNotCancellableError",
     "SubmissionNotFoundError",
+    "TooManyMatchingSubmissionsError",
     "ChunkFailed",
 ]
 
@@ -380,6 +382,9 @@ def lookup_submission_ids_by_strategic_metadata(
         given key-value pairs, but it may also contain other key-value pairs.
 
         Raises:
+        - `TooManyMatchingSubmissionsError` if the lookup matches more
+          submissions than the server's configured maximum. Narrow the query
+          with more specific strategic metadata.
         - `InternalProducerClientError` if there is a low-level internal error.
 
         """
diff --git a/libs/opsqueue_python/src/errors.rs b/libs/opsqueue_python/src/errors.rs
index f0c68508..aafd96e3 100644
--- a/libs/opsqueue_python/src/errors.rs
+++ b/libs/opsqueue_python/src/errors.rs
@@ -5,7 +5,7 @@ use std::error::Error;
 use opsqueue::common::chunk::ChunkId;
 use opsqueue::common::errors::{
     ChunkNotFound, IncorrectUsage, SubmissionNotCancellable, SubmissionNotFound,
-    UnexpectedOpsqueueConsumerServerResponse, E,
+    TooManyMatchingSubmissions, UnexpectedOpsqueueConsumerServerResponse, E,
 };
 use pyo3::exceptions::PyBaseException;
 use pyo3::{import_exception, Bound, PyErr, Python};
@@ -22,6 +22,7 @@ import_exception!(opsqueue.exceptions, TryFromIntError);
 import_exception!(opsqueue.exceptions, ChunkNotFoundError);
 import_exception!(opsqueue.exceptions, SubmissionNotFoundError);
 import_exception!(opsqueue.exceptions, SubmissionNotCancellableError);
+import_exception!(opsqueue.exceptions, TooManyMatchingSubmissionsError);
 import_exception!(opsqueue.exceptions, NewObjectStoreClientError);
 import_exception!(opsqueue.exceptions, SubmissionNotCompletedYetError);
 
@@ -146,6 +147,12 @@ impl From<CError<SubmissionNotFound>> for PyErr {
     }
 }
 
+impl From<CError<TooManyMatchingSubmissions>> for PyErr {
+    fn from(value: CError<TooManyMatchingSubmissions>) -> Self {
+        TooManyMatchingSubmissionsError::new_err(value.0 .0)
+    }
+}
+
 pub struct SubmissionFailed(
     pub crate::common::SubmissionFailed,
     pub crate::common::ChunkFailed,
diff --git a/libs/opsqueue_python/src/producer.rs b/libs/opsqueue_python/src/producer.rs
index 212e9438..59e13e90 100644
--- a/libs/opsqueue_python/src/producer.rs
+++ b/libs/opsqueue_python/src/producer.rs
@@ -10,7 +10,7 @@ use pyo3::{
 use futures::{stream::BoxStream, StreamExt, TryStreamExt};
 use opsqueue::{
     common::errors::E::{self, L, R},
-    common::errors::{SubmissionNotCancellable, SubmissionNotFound},
+    common::errors::{SubmissionNotCancellable, SubmissionNotFound, TooManyMatchingSubmissions},
     object_store::{ChunksStorageError, NewObjectStoreClientError},
     producer::client::{Client as ActualClient, InternalProducerClientError},
 };
@@ -196,7 +196,14 @@ impl ProducerClient {
         &self,
         py: Python<'_>,
         strategic_metadata: StrategicMetadataMap,
-    ) -> CPyResult<Vec<SubmissionId>, E<FatalPythonException, InternalProducerClientError>> {
+    ) -> CPyResult<
+        Vec<SubmissionId>,
+        E![
+            FatalPythonException,
+            TooManyMatchingSubmissions,
+            InternalProducerClientError
+        ],
+    > {
         py.allow_threads(|| {
             self.block_unless_interrupted(async {
                 self.producer_client
diff --git a/libs/opsqueue_python/tests/test_roundtrip.py b/libs/opsqueue_python/tests/test_roundtrip.py
index 9ea4f2fb..a5d0eb67 100644
--- a/libs/opsqueue_python/tests/test_roundtrip.py
+++ b/libs/opsqueue_python/tests/test_roundtrip.py
@@ -4,7 +4,6 @@
 
 from collections.abc import Iterator, Sequence
 from opsqueue.producer import (
-    LookupIdsWithEmptyStrategicMetadataError,
     SubmissionId,
     ProducerClient,
     SubmissionCompleted,
diff --git a/opsqueue/src/common/errors.rs b/opsqueue/src/common/errors.rs
index f30436eb..f48fea34 100644
--- a/opsqueue/src/common/errors.rs
+++ b/opsqueue/src/common/errors.rs
@@ -49,6 +49,10 @@ pub enum SubmissionNotCancellable {
     Cancelled(SubmissionCancelled),
 }
 
+#[derive(Error, Debug, Deserialize, Serialize)]
+#[error("Too many submissions matched the lookup, the maximum is {0:?}")]
+pub struct TooManyMatchingSubmissions(pub u64);
+
 #[derive(Error, Debug)]
 #[error("Unexpected opsqueue consumer server response. This indicates an error inside Opsqueue itself: {0:?}")]
 pub struct UnexpectedOpsqueueConsumerServerResponse(pub SyncServerToClientResponse);
diff --git a/opsqueue/src/common/mod.rs b/opsqueue/src/common/mod.rs
index a6542749..65aadba6 100644
--- a/opsqueue/src/common/mod.rs
+++ b/opsqueue/src/common/mod.rs
@@ -12,3 +12,50 @@ pub mod submission;
 /// consider hashing them and using that hash as MetaStateVal.
 pub type MetaStateVal = i64;
 pub type StrategicMetadataMap = FxHashMap<String, MetaStateVal>;
+
+/// Maximum number of submissions a lookup may return.
+/// Constructor checks: MaxSubmissions + 1 <= i64::MAX;
+#[derive(Debug, Clone, Copy)]
+pub struct MaxSubmissions(u64);
+
+impl MaxSubmissions {
+    pub fn new(value: u64) -> Result<Self, MaxSubmissionsTooLarge> {
+        if value < i64::MAX as u64 {
+            Ok(Self(value))
+        } else {
+            Err(MaxSubmissionsTooLarge(value))
+        }
+    }
+}
+
+impl From<MaxSubmissions> for u64 {
+    fn from(max_submissions: MaxSubmissions) -> u64 {
+        max_submissions.0
+    }
+}
+
+#[derive(Debug, thiserror::Error)]
+#[error("max_submissions value {0} is too large; it must be at most i64::MAX - 1")]
+pub struct MaxSubmissionsTooLarge(pub u64);
+
+impl std::fmt::Display for MaxSubmissions {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum ParseMaxSubmissionsError {
+    #[error(transparent)]
+    NotANumber(#[from] std::num::ParseIntError),
+    #[error(transparent)]
+    TooLarge(#[from] MaxSubmissionsTooLarge),
+}
+
+impl std::str::FromStr for MaxSubmissions {
+    type Err = ParseMaxSubmissionsError;
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        let value: u64 = s.parse()?;
+        Ok(MaxSubmissions::new(value)?)
+    }
+}
diff --git a/opsqueue/src/common/submission.rs b/opsqueue/src/common/submission.rs
index 3f52ac14..a9a37e4c 100644
--- a/opsqueue/src/common/submission.rs
+++ b/opsqueue/src/common/submission.rs
@@ -269,14 +269,17 @@ impl Submission {
 pub mod db {
     use crate::{
         common::{
-            errors::{DatabaseError, SubmissionNotCancellable, SubmissionNotFound, E},
-            StrategicMetadataMap,
+            errors::{
+                DatabaseError, SubmissionNotCancellable, SubmissionNotFound,
+                TooManyMatchingSubmissions, E,
+            },
+            MaxSubmissions, StrategicMetadataMap,
         },
         db::{Connection, True, WriterConnection, WriterPool},
     };
-    use chunk::ChunkSize;
-    use sqlx::{QueryBuilder, Row, Sqlite, query, query_scalar};
     use axum_prometheus::metrics::{counter, histogram};
+    use chunk::ChunkSize;
+    use sqlx::{query, query_scalar, QueryBuilder, Row, Sqlite};
 
     use super::*;
 
@@ -545,18 +548,35 @@ pub mod db {
 
     pub async fn lookup_ids_by_strategic_metadata(
         strategic_metadata: StrategicMetadataMap,
+        max_submissions: MaxSubmissions,
         mut conn: impl Connection,
-    ) -> Result<Vec<SubmissionId>, DatabaseError> {
+    ) -> Result<Vec<SubmissionId>, E<DatabaseError, TooManyMatchingSubmissions>> {
+        // MaxSubmissions provides us with the guarantee this won't overflow.
+        let limit = (u64::from(max_submissions) + 1) as i64;
+
+        // If 'max_submissions' results is exceeded return an explicit error.
+        fn error_if_too_many(
+            max_submissions: MaxSubmissions,
+            ids: Vec<SubmissionId>,
+        ) -> Result<Vec<SubmissionId>, E<DatabaseError, TooManyMatchingSubmissions>> {
+            if ids.len() as u64 > u64::from(max_submissions) {
+                Err(E::R(TooManyMatchingSubmissions(u64::from(max_submissions))))
+            } else {
+                Ok(ids)
+            }
+        }
+
         // The main query below to match on strategic_metadata will fail at
         // run-time if strategic_metadata is empty (because the query will
         // produce `IN ()`). So we handle the empty case here separately.
         if strategic_metadata.is_empty() {
             let ids = query_scalar!(
-                r#"SELECT id AS "id: SubmissionId" FROM submissions ORDER by id"#,
+                r#"SELECT id AS "id: SubmissionId" FROM submissions ORDER by id LIMIT ?"#,
+                limit
             )
             .fetch_all(conn.get_inner())
             .await?;
-            return Ok(ids);
+            return error_if_too_many(max_submissions, ids);
         }
         // SQLx currently only supports "WHERE X IN (a, ...)" queries for postgres:
         // https://github.com/transact-rs/sqlx/blob/main/FAQ.md#how-can-i-do-a-select--where-foo-in--query
@@ -575,12 +595,15 @@ pub mod db {
         });
         query_builder.push(") GROUP BY submission_id HAVING count(*) = ");
         query_builder.push_bind(strategic_metadata.len() as i64);
-        query_builder.push(" ORDER BY submission_id");
+        query_builder.push(" ORDER BY submission_id limit ");
+        query_builder.push_bind(limit);
         let rows = query_builder.build().fetch_all(conn.get_inner()).await?;
-        Ok(rows
-            .into_iter()
-            .map(|row| row.get("submission_id"))
-            .collect())
+        error_if_too_many(
+            max_submissions,
+            rows.into_iter()
+                .map(|row| row.get("submission_id"))
+                .collect(),
+        )
     }
 
     #[tracing::instrument(skip(conn))]
diff --git a/opsqueue/src/config.rs b/opsqueue/src/config.rs
index 08acd428..4af004c2 100644
--- a/opsqueue/src/config.rs
+++ b/opsqueue/src/config.rs
@@ -6,6 +6,12 @@ use std::num::NonZero;
 
 use clap::Parser;
 
+use crate::common::MaxSubmissions;
+
+fn default_max_submissions_returned() -> MaxSubmissions {
+    MaxSubmissions::new(100_000).expect("Valid MaxSubmissions default")
+}
+
 /// Making big work horizontally scalable.
 #[derive(Parser, Debug)]
 #[command(version, about, long_about = None)]
@@ -77,6 +83,11 @@ pub struct Config {
 
     #[arg(long, default_value = "1 hour")]
     pub max_submission_age: humantime::Duration,
+
+    /// Maximum number of submission IDs that a single
+    /// `lookup_submission_ids_by_strategic_metadata` request may return.
+    #[arg(long, default_value_t = default_max_submissions_returned())]
+    pub max_submissions_returned: MaxSubmissions,
 }
 
 impl Default for Config {
@@ -92,6 +103,7 @@ impl Default for Config {
         let max_missable_heartbeats = 3;
         let max_chunk_retries = 10;
         let max_submission_age = humantime::Duration::from_str("1 hour").expect("valid humantime");
+        let max_submissions_returned = default_max_submissions_returned();
         Config {
             port,
             database_filename,
@@ -101,6 +113,7 @@ impl Default for Config {
             max_missable_heartbeats,
             max_chunk_retries,
             max_submission_age,
+            max_submissions_returned,
         }
     }
 }
diff --git a/opsqueue/src/producer/client.rs b/opsqueue/src/producer/client.rs
index 2bfa9052..6bf85036 100644
--- a/opsqueue/src/producer/client.rs
+++ b/opsqueue/src/producer/client.rs
@@ -8,7 +8,7 @@ use http::StatusCode;
 use crate::{
     common::{
         errors::E::{L, R},
-        errors::{SubmissionNotCancellable, SubmissionNotFound},
+        errors::{SubmissionNotCancellable, SubmissionNotFound, TooManyMatchingSubmissions},
         submission::{SubmissionId, SubmissionStatus},
         StrategicMetadataMap,
     },
@@ -230,24 +230,45 @@ impl Client {
     pub async fn lookup_submission_ids_by_strategic_metadata(
         &self,
         strategic_metadata: &StrategicMetadataMap,
-    ) -> Result<Vec<SubmissionId>, InternalProducerClientError> {
+    ) -> Result<Vec<SubmissionId>, E![TooManyMatchingSubmissions, InternalProducerClientError]>
+    {
         (|| async {
             let base_url = &self.base_url;
-            let resp = self
+            let response = self
                 .http_client
                 .post(format!(
                     "{base_url}/submissions/lookup_ids_by_strategic_metadata"
                 ))
                 .json(strategic_metadata)
                 .send()
-                .await?
-                .error_for_status()?;
-            let bytes = resp.bytes().await?;
-            let body = serde_json::from_slice(&bytes)?;
-            Ok(body)
+                .await
+                .map_err(|e| R(e.into()))?;
+            let status = response.status();
+            match status {
+                // 200, the lookup succeeded.
+                StatusCode::OK => {
+                    let submission_ids = response
+                        .json::<Vec<SubmissionId>>()
+                        .await
+                        .map_err(|e| R(e.into()))?;
+                    Ok(submission_ids)
+                }
+                // 400, matched more submissions than the configured maximum.
+                StatusCode::BAD_REQUEST => {
+                    let too_many_err = response
+                        .json::<TooManyMatchingSubmissions>()
+                        .await
+                        .map_err(|e| R(e.into()))?;
+                    Err(L(too_many_err))
+                }
+                _ => Err(R(InternalProducerClientError::UnexpectedStatus(status))),
+            }
         })
         .retry(retry_policy())
-        .when(InternalProducerClientError::is_ephemeral)
+        .when(|e| match e {
+            L(_) => false,
+            R(client_err) => client_err.is_ephemeral(),
+        })
         .notify(|err, dur| {
             tracing::debug!("retrying error {err:?} with sleeping {dur:?}");
         })
diff --git a/opsqueue/src/producer/server.rs b/opsqueue/src/producer/server.rs
index 49865055..2788d071 100644
--- a/opsqueue/src/producer/server.rs
+++ b/opsqueue/src/producer/server.rs
@@ -2,7 +2,7 @@ use std::sync::Arc;
 
 use crate::common::errors::E::{L, R};
 use crate::common::submission::{self, SubmissionId};
-use crate::common::StrategicMetadataMap;
+use crate::common::{MaxSubmissions, StrategicMetadataMap};
 use crate::db::{self, DBPools};
 use axum::extract;
 use axum::extract::{Path, State};
@@ -15,7 +15,8 @@ use tokio::sync::Notify;
 use super::common::{ChunkContents, InsertSubmission};
 
 pub async fn serve_for_tests(database_pool: DBPools, server_addr: Box<str>) {
-    ServerState::new(database_pool, Arc::new(Notify::new()))
+    let max_submissions = crate::config::Config::default().max_submissions_returned;
+    ServerState::new(database_pool, Arc::new(Notify::new()), max_submissions)
         .serve_for_tests(server_addr)
         .await;
 }
@@ -24,13 +25,19 @@ pub async fn serve_for_tests(database_pool: DBPools, server_addr: Box<str>) {
 pub struct ServerState {
     pool: DBPools,
     notify_on_insert: Arc<Notify>,
+    max_submissions: MaxSubmissions,
 }
 
 impl ServerState {
-    pub fn new(pool: DBPools, notify_on_insert: Arc<Notify>) -> Self {
+    pub fn new(
+        pool: DBPools,
+        notify_on_insert: Arc<Notify>,
+        max_submissions: MaxSubmissions,
+    ) -> Self {
         ServerState {
             pool,
             notify_on_insert,
+            max_submissions,
         }
     }
     pub async fn serve_for_tests(self, server_addr: Box<str>) {
@@ -139,14 +146,28 @@ async fn lookup_submission_id_by_prefix(
     Ok(Json(submission_id))
 }
 
+/// 200 if the query was successful.
+/// 400 if the query exceeded the maximum lookup amount.
 async fn lookup_submission_ids_by_strategic_metadata(
     State(state): State<ServerState>,
     extract::Json(strategic_metadata): extract::Json<StrategicMetadataMap>,
-) -> Result<Json<Vec<SubmissionId>>, ServerError> {
-    let mut conn = state.pool.reader_conn().await?;
-    let submission_ids =
-        submission::db::lookup_ids_by_strategic_metadata(strategic_metadata, &mut conn).await?;
-    Ok(Json(submission_ids))
+) -> Result<Json<Vec<SubmissionId>>, Response> {
+    let mut conn = state
+        .pool
+        .reader_conn()
+        .await
+        .map_err(|e| ServerError(e.into()).into_response())?;
+    match submission::db::lookup_ids_by_strategic_metadata(
+        strategic_metadata,
+        state.max_submissions,
+        &mut conn,
+    )
+    .await
+    {
+        Ok(submission_ids) => Ok(Json(submission_ids)),
+        Err(L(db_err)) => Err(ServerError(db_err.into()).into_response()),
+        Err(R(too_many_err)) => Err((StatusCode::BAD_REQUEST, Json(too_many_err)).into_response()),
+    }
 }
 
 #[tracing::instrument(level = "debug", skip(state))]
diff --git a/opsqueue/src/server.rs b/opsqueue/src/server.rs
index cfc9ec4b..58c2bec9 100644
--- a/opsqueue/src/server.rs
+++ b/opsqueue/src/server.rs
@@ -80,8 +80,12 @@ pub fn build_router(
     )
     .run_background()
     .build_router();
-    let producer_routes =
-        crate::producer::server::ServerState::new(pool, notify_on_insert).build_router();
+    let producer_routes = crate::producer::server::ServerState::new(
+        pool,
+        notify_on_insert,
+        config.max_submissions_returned,
+    )
+    .build_router();
 
     let routes = Router::new()
         .nest("/producer", producer_routes)

From d97ef8647172bd07b3d51c8b660c436869d808d3 Mon Sep 17 00:00:00 2001
From: "jeremy.barisch.rooney@channable.com"
 <jeremy.barisch.rooney@channable.com>
Date: Tue, 7 Jul 2026 18:07:26 +0200
Subject: [PATCH 05/16] fixup! fixup! Add
 ProducerClient.lookup_submission_ids_by_strategic_metadata

Test it
---
 justfile                                     |  4 ++--
 libs/opsqueue_python/tests/conftest.py       |  3 ++-
 libs/opsqueue_python/tests/test_roundtrip.py | 24 +++++++++++++++++++-
 3 files changed, 27 insertions(+), 4 deletions(-)

diff --git a/justfile b/justfile
index 65243ab6..cdba3709 100644
--- a/justfile
+++ b/justfile
@@ -47,7 +47,7 @@ test-integration *TEST_ARGS: build-bin build-python
   cd libs/opsqueue_python
   source "./.setup_local_venv.sh"
 
-  timeout 60 pytest --color=yes {{TEST_ARGS}}
+  timeout 600 pytest --color=yes {{TEST_ARGS}}
 
 # Python integration test suite, using artefacts built through Nix. Args are forwarded to pytest
 [group('nix')]
@@ -61,7 +61,7 @@ nix-test-integration *TEST_ARGS: nix-build-bin
   export OPSQUEUE_VIA_NIX=true
   export RUST_LOG="opsqueue=debug"
 
-  timeout 60 pytest --color=yes {{TEST_ARGS}}
+  timeout 600 pytest --color=yes {{TEST_ARGS}}
 
 # Run all linters, fast and slow
 [group('lint')]
diff --git a/libs/opsqueue_python/tests/conftest.py b/libs/opsqueue_python/tests/conftest.py
index 4b0ebb41..e6945ea7 100644
--- a/libs/opsqueue_python/tests/conftest.py
+++ b/libs/opsqueue_python/tests/conftest.py
@@ -53,7 +53,7 @@ def opsqueue() -> Generator[OpsqueueProcess, None, None]:
 
 @contextmanager
 def opsqueue_service(
-    *, port: int | None = None
+        *, port: int | None = None, command_args: Iterable[str] = ()
 ) -> Generator[OpsqueueProcess, None, None]:
     global test_opsqueue_port_offset
 
@@ -75,6 +75,7 @@ def opsqueue_service(
         str(port),
         "--database-filename",
         temp_dbname,
+        *command_args,
     ]
     env = os.environ.copy()  # We copy the env so e.g. RUST_LOG and other env vars are propagated from outside of the invocation of pytest
     if env.get("RUST_LOG") is None:
diff --git a/libs/opsqueue_python/tests/test_roundtrip.py b/libs/opsqueue_python/tests/test_roundtrip.py
index a5d0eb67..4db7636e 100644
--- a/libs/opsqueue_python/tests/test_roundtrip.py
+++ b/libs/opsqueue_python/tests/test_roundtrip.py
@@ -14,6 +14,7 @@
     SubmissionNotFoundError,
     SubmissionNotCancellable,
     SubmissionNotCancellableError,
+    TooManyMatchingSubmissionsError,
 )
 from opsqueue.consumer import ConsumerClient, Chunk
 from opsqueue.common import SerializationFormat
@@ -21,11 +22,11 @@
     background_process,
     multiple_background_processes,
     OpsqueueProcess,
+    opsqueue_service,
     StrategyDescription,
     strategy_from_description,
 )
 import logging
-
 import pytest
 
 
@@ -561,3 +562,24 @@ def test_lookup_submission_ids_by_empty_strategic_metadata(
     for _ in range(count):
         producer_client.insert_submission([1], chunk_size=1)
     assert len(producer_client.lookup_submission_ids_by_strategic_metadata({})) == count
+
+def test_lookup_too_many_submission_ids_by_strategic_metadata() -> None:
+    """Lookup of too many submission IDs beyond the configured limit raises
+    TooManyMatchingSubmissionsError.
+
+    """
+    max_ = 2
+    # We didn't request the OpsQueueProcess as a parameter so an instance isn't
+    # started, instead we start one here with custom args.
+    with opsqueue_service(command_args=["--max-submissions-returned", str(max_)]) as opsqueue:
+        url = "file:///tmp/opsqueue/test_lookup_too_many_matching_submissions"
+        producer_client = ProducerClient(f"localhost:{opsqueue.port}", url)
+        inserted = 0
+        for _ in range(max_ + 1):
+            producer_client.insert_submission(
+                [1], chunk_size=1, strategic_metadata={"k": 1}
+            )
+            inserted += 1
+        with pytest.raises(TooManyMatchingSubmissionsError):
+            assert inserted == max_ + 1 # Make sure Exception wasn't raised too early.
+            producer_client.lookup_submission_ids_by_strategic_metadata({"k": 1})

From 335312ab573b044d78713ec6b7a6a0961c4e11e8 Mon Sep 17 00:00:00 2001
From: "jeremy.barisch.rooney@channable.com"
 <jeremy.barisch.rooney@channable.com>
Date: Thu, 9 Jul 2026 18:31:54 +0200
Subject: [PATCH 06/16] fixup! Add
 ProducerClient.lookup_submission_ids_by_strategic_metadata

assert query plan
---
 justfile                                      |  4 +-
 libs/opsqueue_python/tests/conftest.py        |  2 +-
 libs/opsqueue_python/tests/test_roundtrip.py  |  7 +-
 opsqueue/src/common/submission.rs             | 85 ++++++++++++++-----
 opsqueue/src/consumer/dispatcher/metastate.rs |  1 +
 5 files changed, 72 insertions(+), 27 deletions(-)

diff --git a/justfile b/justfile
index cdba3709..65243ab6 100644
--- a/justfile
+++ b/justfile
@@ -47,7 +47,7 @@ test-integration *TEST_ARGS: build-bin build-python
   cd libs/opsqueue_python
   source "./.setup_local_venv.sh"
 
-  timeout 600 pytest --color=yes {{TEST_ARGS}}
+  timeout 60 pytest --color=yes {{TEST_ARGS}}
 
 # Python integration test suite, using artefacts built through Nix. Args are forwarded to pytest
 [group('nix')]
@@ -61,7 +61,7 @@ nix-test-integration *TEST_ARGS: nix-build-bin
   export OPSQUEUE_VIA_NIX=true
   export RUST_LOG="opsqueue=debug"
 
-  timeout 600 pytest --color=yes {{TEST_ARGS}}
+  timeout 60 pytest --color=yes {{TEST_ARGS}}
 
 # Run all linters, fast and slow
 [group('lint')]
diff --git a/libs/opsqueue_python/tests/conftest.py b/libs/opsqueue_python/tests/conftest.py
index e6945ea7..32ec8c68 100644
--- a/libs/opsqueue_python/tests/conftest.py
+++ b/libs/opsqueue_python/tests/conftest.py
@@ -53,7 +53,7 @@ def opsqueue() -> Generator[OpsqueueProcess, None, None]:
 
 @contextmanager
 def opsqueue_service(
-        *, port: int | None = None, command_args: Iterable[str] = ()
+    *, port: int | None = None, command_args: Iterable[str] = ()
 ) -> Generator[OpsqueueProcess, None, None]:
     global test_opsqueue_port_offset
 
diff --git a/libs/opsqueue_python/tests/test_roundtrip.py b/libs/opsqueue_python/tests/test_roundtrip.py
index 4db7636e..3963162d 100644
--- a/libs/opsqueue_python/tests/test_roundtrip.py
+++ b/libs/opsqueue_python/tests/test_roundtrip.py
@@ -563,6 +563,7 @@ def test_lookup_submission_ids_by_empty_strategic_metadata(
         producer_client.insert_submission([1], chunk_size=1)
     assert len(producer_client.lookup_submission_ids_by_strategic_metadata({})) == count
 
+
 def test_lookup_too_many_submission_ids_by_strategic_metadata() -> None:
     """Lookup of too many submission IDs beyond the configured limit raises
     TooManyMatchingSubmissionsError.
@@ -571,7 +572,9 @@ def test_lookup_too_many_submission_ids_by_strategic_metadata() -> None:
     max_ = 2
     # We didn't request the OpsQueueProcess as a parameter so an instance isn't
     # started, instead we start one here with custom args.
-    with opsqueue_service(command_args=["--max-submissions-returned", str(max_)]) as opsqueue:
+    with opsqueue_service(
+        command_args=["--max-submissions-returned", str(max_)]
+    ) as opsqueue:
         url = "file:///tmp/opsqueue/test_lookup_too_many_matching_submissions"
         producer_client = ProducerClient(f"localhost:{opsqueue.port}", url)
         inserted = 0
@@ -581,5 +584,5 @@ def test_lookup_too_many_submission_ids_by_strategic_metadata() -> None:
             )
             inserted += 1
         with pytest.raises(TooManyMatchingSubmissionsError):
-            assert inserted == max_ + 1 # Make sure Exception wasn't raised too early.
+            assert inserted == max_ + 1  # Make sure Exception wasn't raised too early.
             producer_client.lookup_submission_ids_by_strategic_metadata({"k": 1})
diff --git a/opsqueue/src/common/submission.rs b/opsqueue/src/common/submission.rs
index a9a37e4c..43047b4d 100644
--- a/opsqueue/src/common/submission.rs
+++ b/opsqueue/src/common/submission.rs
@@ -566,9 +566,8 @@ pub mod db {
             }
         }
 
-        // The main query below to match on strategic_metadata will fail at
-        // run-time if strategic_metadata is empty (because the query will
-        // produce `IN ()`). So we handle the empty case here separately.
+        // The main query to match on strategic_metadata will fail at run-time
+        // if strategic_metadata is empty, so we handle the empty case here.
         if strategic_metadata.is_empty() {
             let ids = query_scalar!(
                 r#"SELECT id AS "id: SubmissionId" FROM submissions ORDER by id LIMIT ?"#,
@@ -578,25 +577,8 @@ pub mod db {
             .await?;
             return error_if_too_many(max_submissions, ids);
         }
-        // SQLx currently only supports "WHERE X IN (a, ...)" queries for postgres:
-        // https://github.com/transact-rs/sqlx/blob/main/FAQ.md#how-can-i-do-a-select--where-foo-in--query
-        // So we workaround this by manually building the query, foregoing
-        // sqlx's nice type-checking.
-        let mut query_builder: QueryBuilder<Sqlite> = QueryBuilder::new(
-            "
-            SELECT submission_id
-            FROM submissions_metadata
-            INNER JOIN submissions on submissions.id = submission_id
-            WHERE (metadata_key, metadata_value) IN (
-        ",
-        );
-        query_builder.push_values(strategic_metadata.iter(), |mut b, sm| {
-            b.push_bind(sm.0).push_bind(sm.1);
-        });
-        query_builder.push(") GROUP BY submission_id HAVING count(*) = ");
-        query_builder.push_bind(strategic_metadata.len() as i64);
-        query_builder.push(" ORDER BY submission_id limit ");
-        query_builder.push_bind(limit);
+        let mut query_builder: QueryBuilder<Sqlite> =
+            lookup_ids_by_strategic_metadata_query(&strategic_metadata, limit);
         let rows = query_builder.build().fetch_all(conn.get_inner()).await?;
         error_if_too_many(
             max_submissions,
@@ -606,6 +588,32 @@ pub mod db {
         )
     }
 
+    /// The query in 'lookup_ids_by_strategic_metadata', extracted for testing.
+    pub fn lookup_ids_by_strategic_metadata_query(
+        strategic_metadata: &StrategicMetadataMap,
+        limit: i64,
+    ) -> QueryBuilder<'_, Sqlite> {
+        let mut query_builder: QueryBuilder<Sqlite> =
+            QueryBuilder::new("SELECT s0.submission_id FROM submissions_metadata as s0");
+        // Inner join for each piece of strategic metadata.
+        for i in 1..strategic_metadata.len() {
+            query_builder.push(format!(
+                " JOIN submissions_metadata AS s{i} ON s{i}.submission_id = s0.submission_id"
+            ));
+        }
+        query_builder.push(" INNER JOIN submissions on submissions.id = s0.submission_id");
+        for (i, (key, value)) in strategic_metadata.iter().enumerate() {
+            query_builder.push(if i == 0 { " WHERE" } else { " AND" });
+            query_builder.push(format!(" s{i}.metadata_key = "));
+            query_builder.push_bind(key.clone());
+            query_builder.push(format!(" AND s{i}.metadata_value = "));
+            query_builder.push_bind(*value);
+        }
+        query_builder.push(" ORDER BY s0.submission_id LIMIT ");
+        query_builder.push_bind(limit);
+        query_builder
+    }
+
     #[tracing::instrument(skip(conn))]
     pub async fn submission_status(
         id: SubmissionId,
@@ -1097,6 +1105,7 @@ pub mod test {
     use chrono::Utc;
     use chunk::ChunkSize;
     use itertools::Itertools;
+    use sqlformat::{format, FormatOptions, QueryParams};
     use sqlx::{Row, SqliteConnection};
 
     use crate::common::StrategicMetadataMap;
@@ -1131,6 +1140,38 @@ pub mod test {
         );
     }
 
+    #[sqlx::test]
+    pub async fn test_query_plan_lookup_by_strategic_metadata(db: sqlx::SqlitePool) {
+        let mut conn = db.acquire().await.unwrap();
+        let strategic_metadata: StrategicMetadataMap =
+            [("company_id".to_string(), 1), ("company_id".to_string(), 2)]
+                .into_iter()
+                .collect();
+        let qb = lookup_ids_by_strategic_metadata_query(&strategic_metadata, 100_000);
+        let options = FormatOptions::default();
+        let formatted_query = format(qb.sql(), &QueryParams::None, &options);
+        insta::assert_snapshot!(formatted_query, @"
+        SELECT
+          s0.submission_id
+        FROM
+          submissions_metadata as s0
+          INNER JOIN submissions on submissions.id = s0.submission_id
+        WHERE
+          s0.metadata_key = ?
+          AND s0.metadata_value = ?
+        ORDER BY
+          s0.submission_id
+        LIMIT
+          ?
+        ");
+        let explained = explain_query_plan(&formatted_query, &mut conn).await;
+        assert_non_regressing_query_plan(&formatted_query, &explained);
+        insta::assert_snapshot!(explained, @"
+        7, 0, SEARCH s0 USING COVERING INDEX lookup_submission_by_metadata (metadata_key=? AND metadata_value=?)
+        15, 0, SEARCH submissions USING COVERING INDEX sqlite_autoindex_submissions_1 (id=?)
+        ");
+    }
+
     #[sqlx::test]
     pub async fn test_query_plan_submission_status_in_progress(db: sqlx::SqlitePool) {
         let mut conn = db.acquire().await.unwrap();
diff --git a/opsqueue/src/consumer/dispatcher/metastate.rs b/opsqueue/src/consumer/dispatcher/metastate.rs
index 8e743a3f..a3a6668b 100644
--- a/opsqueue/src/consumer/dispatcher/metastate.rs
+++ b/opsqueue/src/consumer/dispatcher/metastate.rs
@@ -112,6 +112,7 @@ impl MetaStateField {
     pub fn too_high_counts(&self, max: usize) -> impl Iterator<Item = MetaStateVal> + '_ {
         tracing::debug!("metastate: {self:?}");
         self.counts_to_vals
+            // Negatives ?
             .range((max, 0)..)
             .map(|entry| entry.value().1)
     }

From 9a85c035304a3d9f28ce6dfe9bc72ca366d2b066 Mon Sep 17 00:00:00 2001
From: "jeremy.barisch.rooney@channable.com"
 <jeremy.barisch.rooney@channable.com>
Date: Tue, 14 Jul 2026 14:14:59 +0200
Subject: [PATCH 07/16] fixup! Add
 ProducerClient.lookup_submission_ids_by_strategic_metadata

Revert timeout
---
 justfile | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/justfile b/justfile
index 65243ab6..cdba3709 100644
--- a/justfile
+++ b/justfile
@@ -47,7 +47,7 @@ test-integration *TEST_ARGS: build-bin build-python
   cd libs/opsqueue_python
   source "./.setup_local_venv.sh"
 
-  timeout 60 pytest --color=yes {{TEST_ARGS}}
+  timeout 600 pytest --color=yes {{TEST_ARGS}}
 
 # Python integration test suite, using artefacts built through Nix. Args are forwarded to pytest
 [group('nix')]
@@ -61,7 +61,7 @@ nix-test-integration *TEST_ARGS: nix-build-bin
   export OPSQUEUE_VIA_NIX=true
   export RUST_LOG="opsqueue=debug"
 
-  timeout 60 pytest --color=yes {{TEST_ARGS}}
+  timeout 600 pytest --color=yes {{TEST_ARGS}}
 
 # Run all linters, fast and slow
 [group('lint')]

From e09a1ea1e334997295ce7f116b7744e10966bc64 Mon Sep 17 00:00:00 2001
From: "jeremy.barisch.rooney@channable.com"
 <jeremy.barisch.rooney@channable.com>
Date: Tue, 14 Jul 2026 14:15:39 +0200
Subject: [PATCH 08/16] fixup! Add
 ProducerClient.lookup_submission_ids_by_strategic_metadata

rm space
---
 libs/opsqueue_python/src/errors.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/libs/opsqueue_python/src/errors.rs b/libs/opsqueue_python/src/errors.rs
index aafd96e3..1d332633 100644
--- a/libs/opsqueue_python/src/errors.rs
+++ b/libs/opsqueue_python/src/errors.rs
@@ -149,7 +149,7 @@ impl From<CError<SubmissionNotFound>> for PyErr {
 
 impl From<CError<TooManyMatchingSubmissions>> for PyErr {
     fn from(value: CError<TooManyMatchingSubmissions>) -> Self {
-        TooManyMatchingSubmissionsError::new_err(value.0 .0)
+        TooManyMatchingSubmissionsError::new_err(value.0.0)
     }
 }
 

From 6d27f09a6ca98f6cbfdc09224f9570d17d435e96 Mon Sep 17 00:00:00 2001
From: "jeremy.barisch.rooney@channable.com"
 <jeremy.barisch.rooney@channable.com>
Date: Tue, 14 Jul 2026 14:16:11 +0200
Subject: [PATCH 09/16] fixup! Add
 ProducerClient.lookup_submission_ids_by_strategic_metadata

rm LookupIdsWithEmptyStrategicMetadataError
---
 libs/opsqueue_python/python/opsqueue/producer.py | 4 ----
 1 file changed, 4 deletions(-)

diff --git a/libs/opsqueue_python/python/opsqueue/producer.py b/libs/opsqueue_python/python/opsqueue/producer.py
index 7a93da5e..d72f0cd2 100644
--- a/libs/opsqueue_python/python/opsqueue/producer.py
+++ b/libs/opsqueue_python/python/opsqueue/producer.py
@@ -44,10 +44,6 @@
 ]
 
 
-class LookupIdsWithEmptyStrategicMetadataError(Exception):
-    pass
-
-
 class ProducerClient:
     """
     Opsqueue producer client. Allows sending of large collections of operations ('submissions')

From 441d2ca6e722d6ef9a5f4e06992dc3f92405d393 Mon Sep 17 00:00:00 2001
From: "jeremy.barisch.rooney@channable.com"
 <jeremy.barisch.rooney@channable.com>
Date: Tue, 14 Jul 2026 14:46:02 +0200
Subject: [PATCH 10/16] fixup! Add
 ProducerClient.lookup_submission_ids_by_strategic_metadata

Use NonZero<u64>
---
 opsqueue/src/common/mod.rs | 15 ++++++++-------
 opsqueue/src/config.rs     |  2 +-
 2 files changed, 9 insertions(+), 8 deletions(-)

diff --git a/opsqueue/src/common/mod.rs b/opsqueue/src/common/mod.rs
index 65aadba6..503d8ce3 100644
--- a/opsqueue/src/common/mod.rs
+++ b/opsqueue/src/common/mod.rs
@@ -1,5 +1,6 @@
 //! Common datatypes and errors shared across all parts of Opsqueue
 use rustc_hash::FxHashMap;
+use std::num::{NonZero};
 
 pub mod chunk;
 pub mod errors;
@@ -14,13 +15,13 @@ pub type MetaStateVal = i64;
 pub type StrategicMetadataMap = FxHashMap<String, MetaStateVal>;
 
 /// Maximum number of submissions a lookup may return.
-/// Constructor checks: MaxSubmissions + 1 <= i64::MAX;
+/// Guarantees: 0 < MaxSubmissions 1 < i64::MAX;
 #[derive(Debug, Clone, Copy)]
-pub struct MaxSubmissions(u64);
+pub struct MaxSubmissions(NonZero<u64>);
 
 impl MaxSubmissions {
-    pub fn new(value: u64) -> Result<Self, MaxSubmissionsTooLarge> {
-        if value < i64::MAX as u64 {
+    pub fn new(value: NonZero<u64>) -> Result<Self, MaxSubmissionsTooLarge> {
+        if u64::from(value) < i64::MAX as u64 {
             Ok(Self(value))
         } else {
             Err(MaxSubmissionsTooLarge(value))
@@ -30,13 +31,13 @@ impl MaxSubmissions {
 
 impl From<MaxSubmissions> for u64 {
     fn from(max_submissions: MaxSubmissions) -> u64 {
-        max_submissions.0
+        u64::from(max_submissions.0)
     }
 }
 
 #[derive(Debug, thiserror::Error)]
 #[error("max_submissions value {0} is too large; it must be at most i64::MAX - 1")]
-pub struct MaxSubmissionsTooLarge(pub u64);
+pub struct MaxSubmissionsTooLarge(pub NonZero<u64>);
 
 impl std::fmt::Display for MaxSubmissions {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
@@ -55,7 +56,7 @@ pub enum ParseMaxSubmissionsError {
 impl std::str::FromStr for MaxSubmissions {
     type Err = ParseMaxSubmissionsError;
     fn from_str(s: &str) -> Result<Self, Self::Err> {
-        let value: u64 = s.parse()?;
+        let value: NonZero<u64> = s.parse()?;
         Ok(MaxSubmissions::new(value)?)
     }
 }
diff --git a/opsqueue/src/config.rs b/opsqueue/src/config.rs
index 4af004c2..36af92bc 100644
--- a/opsqueue/src/config.rs
+++ b/opsqueue/src/config.rs
@@ -9,7 +9,7 @@ use clap::Parser;
 use crate::common::MaxSubmissions;
 
 fn default_max_submissions_returned() -> MaxSubmissions {
-    MaxSubmissions::new(100_000).expect("Valid MaxSubmissions default")
+    MaxSubmissions::new(NonZero::new(100_000).expect("Non-zero u64")).expect("Valid MaxSubmissions default")
 }
 
 /// Making big work horizontally scalable.

From e84bbc449c17b0b30a2c642ee1d65f688018173f Mon Sep 17 00:00:00 2001
From: "jeremy.barisch.rooney@channable.com"
 <jeremy.barisch.rooney@channable.com>
Date: Tue, 14 Jul 2026 14:53:37 +0200
Subject: [PATCH 11/16] fixup! Add
 ProducerClient.lookup_submission_ids_by_strategic_metadata

Call lookup_submission_ids_by_strategic_metadata every test iteration
---
 libs/opsqueue_python/tests/test_roundtrip.py | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/libs/opsqueue_python/tests/test_roundtrip.py b/libs/opsqueue_python/tests/test_roundtrip.py
index 3963162d..e9393eca 100644
--- a/libs/opsqueue_python/tests/test_roundtrip.py
+++ b/libs/opsqueue_python/tests/test_roundtrip.py
@@ -577,12 +577,12 @@ def test_lookup_too_many_submission_ids_by_strategic_metadata() -> None:
     ) as opsqueue:
         url = "file:///tmp/opsqueue/test_lookup_too_many_matching_submissions"
         producer_client = ProducerClient(f"localhost:{opsqueue.port}", url)
-        inserted = 0
+        inserted = []
         for _ in range(max_ + 1):
-            producer_client.insert_submission(
+            assert inserted == producer_client.lookup_submission_ids_by_strategic_metadata({"k": 1})
+            inserted.append(producer_client.insert_submission(
                 [1], chunk_size=1, strategic_metadata={"k": 1}
-            )
-            inserted += 1
+            ))
         with pytest.raises(TooManyMatchingSubmissionsError):
-            assert inserted == max_ + 1  # Make sure Exception wasn't raised too early.
+            assert len(inserted) == max_ + 1  # Make sure Exception wasn't raised too early.
             producer_client.lookup_submission_ids_by_strategic_metadata({"k": 1})

From cbb1dd0948df0cc0771d57975268ab0170190777 Mon Sep 17 00:00:00 2001
From: "jeremy.barisch.rooney@channable.com"
 <jeremy.barisch.rooney@channable.com>
Date: Tue, 14 Jul 2026 15:13:56 +0200
Subject: [PATCH 12/16] fixup! Add
 ProducerClient.lookup_submission_ids_by_strategic_metadata

Simplify query logic
---
 opsqueue/src/common/submission.rs | 40 ++++++++++---------------------
 1 file changed, 13 insertions(+), 27 deletions(-)

diff --git a/opsqueue/src/common/submission.rs b/opsqueue/src/common/submission.rs
index 43047b4d..26b61984 100644
--- a/opsqueue/src/common/submission.rs
+++ b/opsqueue/src/common/submission.rs
@@ -279,7 +279,7 @@ pub mod db {
     };
     use axum_prometheus::metrics::{counter, histogram};
     use chunk::ChunkSize;
-    use sqlx::{query, query_scalar, QueryBuilder, Row, Sqlite};
+    use sqlx::{query, query_scalar, QueryBuilder, Sqlite};
 
     use super::*;
 
@@ -553,39 +553,25 @@ pub mod db {
     ) -> Result<Vec<SubmissionId>, E<DatabaseError, TooManyMatchingSubmissions>> {
         // MaxSubmissions provides us with the guarantee this won't overflow.
         let limit = (u64::from(max_submissions) + 1) as i64;
-
-        // If 'max_submissions' results is exceeded return an explicit error.
-        fn error_if_too_many(
-            max_submissions: MaxSubmissions,
-            ids: Vec<SubmissionId>,
-        ) -> Result<Vec<SubmissionId>, E<DatabaseError, TooManyMatchingSubmissions>> {
-            if ids.len() as u64 > u64::from(max_submissions) {
-                Err(E::R(TooManyMatchingSubmissions(u64::from(max_submissions))))
-            } else {
-                Ok(ids)
-            }
-        }
-
         // The main query to match on strategic_metadata will fail at run-time
         // if strategic_metadata is empty, so we handle the empty case here.
-        if strategic_metadata.is_empty() {
-            let ids = query_scalar!(
+        let ids = if strategic_metadata.is_empty() {
+            query_scalar!(
                 r#"SELECT id AS "id: SubmissionId" FROM submissions ORDER by id LIMIT ?"#,
                 limit
             )
             .fetch_all(conn.get_inner())
-            .await?;
-            return error_if_too_many(max_submissions, ids);
+            .await?
+        } else {
+            let mut query_builder: QueryBuilder<Sqlite> =
+                lookup_ids_by_strategic_metadata_query(&strategic_metadata, limit);
+            query_builder.build_query_scalar().fetch_all(conn.get_inner()).await?
+        };
+        if ids.len() as u64 > u64::from(max_submissions) {
+            Err(E::R(TooManyMatchingSubmissions(u64::from(max_submissions))))
+        } else {
+            Ok(ids)
         }
-        let mut query_builder: QueryBuilder<Sqlite> =
-            lookup_ids_by_strategic_metadata_query(&strategic_metadata, limit);
-        let rows = query_builder.build().fetch_all(conn.get_inner()).await?;
-        error_if_too_many(
-            max_submissions,
-            rows.into_iter()
-                .map(|row| row.get("submission_id"))
-                .collect(),
-        )
     }
 
     /// The query in 'lookup_ids_by_strategic_metadata', extracted for testing.

From 14a6a7862378e24bd63f7ca769dfcd2d9116b399 Mon Sep 17 00:00:00 2001
From: "jeremy.barisch.rooney@channable.com"
 <jeremy.barisch.rooney@channable.com>
Date: Tue, 14 Jul 2026 15:32:31 +0200
Subject: [PATCH 13/16] fixup! Add
 ProducerClient.lookup_submission_ids_by_strategic_metadata

pre-commit
---
 libs/opsqueue_python/src/errors.rs           |  2 +-
 libs/opsqueue_python/tests/test_roundtrip.py | 19 +++++++++++++------
 opsqueue/src/common/mod.rs                   |  2 +-
 opsqueue/src/common/submission.rs            |  5 ++++-
 opsqueue/src/config.rs                       |  3 ++-
 5 files changed, 21 insertions(+), 10 deletions(-)

diff --git a/libs/opsqueue_python/src/errors.rs b/libs/opsqueue_python/src/errors.rs
index 1d332633..aafd96e3 100644
--- a/libs/opsqueue_python/src/errors.rs
+++ b/libs/opsqueue_python/src/errors.rs
@@ -149,7 +149,7 @@ impl From<CError<SubmissionNotFound>> for PyErr {
 
 impl From<CError<TooManyMatchingSubmissions>> for PyErr {
     fn from(value: CError<TooManyMatchingSubmissions>) -> Self {
-        TooManyMatchingSubmissionsError::new_err(value.0.0)
+        TooManyMatchingSubmissionsError::new_err(value.0 .0)
     }
 }
 
diff --git a/libs/opsqueue_python/tests/test_roundtrip.py b/libs/opsqueue_python/tests/test_roundtrip.py
index e9393eca..0073627b 100644
--- a/libs/opsqueue_python/tests/test_roundtrip.py
+++ b/libs/opsqueue_python/tests/test_roundtrip.py
@@ -577,12 +577,19 @@ def test_lookup_too_many_submission_ids_by_strategic_metadata() -> None:
     ) as opsqueue:
         url = "file:///tmp/opsqueue/test_lookup_too_many_matching_submissions"
         producer_client = ProducerClient(f"localhost:{opsqueue.port}", url)
-        inserted = []
+        inserted: list[SubmissionId] = []
         for _ in range(max_ + 1):
-            assert inserted == producer_client.lookup_submission_ids_by_strategic_metadata({"k": 1})
-            inserted.append(producer_client.insert_submission(
-                [1], chunk_size=1, strategic_metadata={"k": 1}
-            ))
+            assert (
+                inserted
+                == producer_client.lookup_submission_ids_by_strategic_metadata({"k": 1})
+            )
+            inserted.append(
+                producer_client.insert_submission(
+                    [1], chunk_size=1, strategic_metadata={"k": 1}
+                )
+            )
         with pytest.raises(TooManyMatchingSubmissionsError):
-            assert len(inserted) == max_ + 1  # Make sure Exception wasn't raised too early.
+            assert (
+                len(inserted) == max_ + 1
+            )  # Make sure Exception wasn't raised too early.
             producer_client.lookup_submission_ids_by_strategic_metadata({"k": 1})
diff --git a/opsqueue/src/common/mod.rs b/opsqueue/src/common/mod.rs
index 503d8ce3..4a305098 100644
--- a/opsqueue/src/common/mod.rs
+++ b/opsqueue/src/common/mod.rs
@@ -1,6 +1,6 @@
 //! Common datatypes and errors shared across all parts of Opsqueue
 use rustc_hash::FxHashMap;
-use std::num::{NonZero};
+use std::num::NonZero;
 
 pub mod chunk;
 pub mod errors;
diff --git a/opsqueue/src/common/submission.rs b/opsqueue/src/common/submission.rs
index 26b61984..1549e015 100644
--- a/opsqueue/src/common/submission.rs
+++ b/opsqueue/src/common/submission.rs
@@ -565,7 +565,10 @@ pub mod db {
         } else {
             let mut query_builder: QueryBuilder<Sqlite> =
                 lookup_ids_by_strategic_metadata_query(&strategic_metadata, limit);
-            query_builder.build_query_scalar().fetch_all(conn.get_inner()).await?
+            query_builder
+                .build_query_scalar()
+                .fetch_all(conn.get_inner())
+                .await?
         };
         if ids.len() as u64 > u64::from(max_submissions) {
             Err(E::R(TooManyMatchingSubmissions(u64::from(max_submissions))))
diff --git a/opsqueue/src/config.rs b/opsqueue/src/config.rs
index 36af92bc..c899d2b3 100644
--- a/opsqueue/src/config.rs
+++ b/opsqueue/src/config.rs
@@ -9,7 +9,8 @@ use clap::Parser;
 use crate::common::MaxSubmissions;
 
 fn default_max_submissions_returned() -> MaxSubmissions {
-    MaxSubmissions::new(NonZero::new(100_000).expect("Non-zero u64")).expect("Valid MaxSubmissions default")
+    MaxSubmissions::new(NonZero::new(100_000).expect("Non-zero u64"))
+        .expect("Valid MaxSubmissions default")
 }
 
 /// Making big work horizontally scalable.

From 76d4ede5cb9808e0a76c292ba2f4423fc3ef7030 Mon Sep 17 00:00:00 2001
From: "jeremy.barisch.rooney@channable.com"
 <jeremy.barisch.rooney@channable.com>
Date: Wed, 15 Jul 2026 10:21:19 +0200
Subject: [PATCH 14/16] fixup! Add
 ProducerClient.lookup_submission_ids_by_strategic_metadata

address comments
---
 libs/opsqueue_python/tests/test_roundtrip.py | 17 +++++++------
 opsqueue/src/common/submission.rs            | 25 +++++++++++---------
 2 files changed, 22 insertions(+), 20 deletions(-)

diff --git a/libs/opsqueue_python/tests/test_roundtrip.py b/libs/opsqueue_python/tests/test_roundtrip.py
index 0073627b..2251c329 100644
--- a/libs/opsqueue_python/tests/test_roundtrip.py
+++ b/libs/opsqueue_python/tests/test_roundtrip.py
@@ -578,18 +578,17 @@ def test_lookup_too_many_submission_ids_by_strategic_metadata() -> None:
         url = "file:///tmp/opsqueue/test_lookup_too_many_matching_submissions"
         producer_client = ProducerClient(f"localhost:{opsqueue.port}", url)
         inserted: list[SubmissionId] = []
+        strategic_metadata = {"k": 1}
         for _ in range(max_ + 1):
             assert (
                 inserted
-                == producer_client.lookup_submission_ids_by_strategic_metadata({"k": 1})
+                == producer_client.lookup_submission_ids_by_strategic_metadata(strategic_metadata)
             )
             inserted.append(
-                producer_client.insert_submission(
-                    [1], chunk_size=1, strategic_metadata={"k": 1}
-                )
+                producer_client.insert_submission([1], chunk_size=1, strategic_metadata=strategic_metadata)
             )
-        with pytest.raises(TooManyMatchingSubmissionsError):
-            assert (
-                len(inserted) == max_ + 1
-            )  # Make sure Exception wasn't raised too early.
-            producer_client.lookup_submission_ids_by_strategic_metadata({"k": 1})
+        with pytest.raises(TooManyMatchingSubmissionsError) as exc:
+            assert len(inserted) == max_ + 1
+            producer_client.lookup_submission_ids_by_strategic_metadata(strategic_metadata)
+        assert exc.type is TooManyMatchingSubmissionsError
+        assert exc.value.max_submissions == max_
diff --git a/opsqueue/src/common/submission.rs b/opsqueue/src/common/submission.rs
index 1549e015..c71763c6 100644
--- a/opsqueue/src/common/submission.rs
+++ b/opsqueue/src/common/submission.rs
@@ -557,7 +557,7 @@ pub mod db {
         // if strategic_metadata is empty, so we handle the empty case here.
         let ids = if strategic_metadata.is_empty() {
             query_scalar!(
-                r#"SELECT id AS "id: SubmissionId" FROM submissions ORDER by id LIMIT ?"#,
+                r#"SELECT id AS "id: SubmissionId" FROM submissions ORDER BY id LIMIT ?"#,
                 limit
             )
             .fetch_all(conn.get_inner())
@@ -583,14 +583,13 @@ pub mod db {
         limit: i64,
     ) -> QueryBuilder<'_, Sqlite> {
         let mut query_builder: QueryBuilder<Sqlite> =
-            QueryBuilder::new("SELECT s0.submission_id FROM submissions_metadata as s0");
+            QueryBuilder::new("SELECT id FROM submissions");
         // Inner join for each piece of strategic metadata.
-        for i in 1..strategic_metadata.len() {
+        for i in 0..strategic_metadata.len() {
             query_builder.push(format!(
-                " JOIN submissions_metadata AS s{i} ON s{i}.submission_id = s0.submission_id"
+                " JOIN submissions_metadata AS s{i} ON s{i}.submission_id = submissions.id"
             ));
         }
-        query_builder.push(" INNER JOIN submissions on submissions.id = s0.submission_id");
         for (i, (key, value)) in strategic_metadata.iter().enumerate() {
             query_builder.push(if i == 0 { " WHERE" } else { " AND" });
             query_builder.push(format!(" s{i}.metadata_key = "));
@@ -1133,7 +1132,7 @@ pub mod test {
     pub async fn test_query_plan_lookup_by_strategic_metadata(db: sqlx::SqlitePool) {
         let mut conn = db.acquire().await.unwrap();
         let strategic_metadata: StrategicMetadataMap =
-            [("company_id".to_string(), 1), ("company_id".to_string(), 2)]
+            [("company_id".to_string(), 1), ("project_id".to_string(), 2)]
                 .into_iter()
                 .collect();
         let qb = lookup_ids_by_strategic_metadata_query(&strategic_metadata, 100_000);
@@ -1141,13 +1140,16 @@ pub mod test {
         let formatted_query = format(qb.sql(), &QueryParams::None, &options);
         insta::assert_snapshot!(formatted_query, @"
         SELECT
-          s0.submission_id
+          id
         FROM
-          submissions_metadata as s0
-          INNER JOIN submissions on submissions.id = s0.submission_id
+          submissions
+          JOIN submissions_metadata AS s0 ON s0.submission_id = submissions.id
+          JOIN submissions_metadata AS s1 ON s1.submission_id = submissions.id
         WHERE
           s0.metadata_key = ?
           AND s0.metadata_value = ?
+          AND s1.metadata_key = ?
+          AND s1.metadata_value = ?
         ORDER BY
           s0.submission_id
         LIMIT
@@ -1156,8 +1158,9 @@ pub mod test {
         let explained = explain_query_plan(&formatted_query, &mut conn).await;
         assert_non_regressing_query_plan(&formatted_query, &explained);
         insta::assert_snapshot!(explained, @"
-        7, 0, SEARCH s0 USING COVERING INDEX lookup_submission_by_metadata (metadata_key=? AND metadata_value=?)
-        15, 0, SEARCH submissions USING COVERING INDEX sqlite_autoindex_submissions_1 (id=?)
+        8, 0, SEARCH s0 USING COVERING INDEX lookup_submission_by_metadata (metadata_key=? AND metadata_value=?)
+        16, 0, SEARCH submissions USING COVERING INDEX sqlite_autoindex_submissions_1 (id=?)
+        21, 0, SEARCH s1 USING PRIMARY KEY (submission_id=? AND metadata_key=? AND metadata_value=?)
         ");
     }
 

From 23ed08c1b042804c394ddd0f5ae82f0f5494687f Mon Sep 17 00:00:00 2001
From: "jeremy.barisch.rooney@channable.com"
 <jeremy.barisch.rooney@channable.com>
Date: Wed, 15 Jul 2026 13:56:38 +0200
Subject: [PATCH 15/16] fixup! Add
 ProducerClient.lookup_submission_ids_by_strategic_metadata

even nicer sql
---
 libs/opsqueue_python/tests/test_roundtrip.py | 12 +++++++++---
 opsqueue/src/common/submission.rs            | 16 ++++++----------
 2 files changed, 15 insertions(+), 13 deletions(-)

diff --git a/libs/opsqueue_python/tests/test_roundtrip.py b/libs/opsqueue_python/tests/test_roundtrip.py
index 2251c329..1f6ac50c 100644
--- a/libs/opsqueue_python/tests/test_roundtrip.py
+++ b/libs/opsqueue_python/tests/test_roundtrip.py
@@ -582,13 +582,19 @@ def test_lookup_too_many_submission_ids_by_strategic_metadata() -> None:
         for _ in range(max_ + 1):
             assert (
                 inserted
-                == producer_client.lookup_submission_ids_by_strategic_metadata(strategic_metadata)
+                == producer_client.lookup_submission_ids_by_strategic_metadata(
+                    strategic_metadata
+                )
             )
             inserted.append(
-                producer_client.insert_submission([1], chunk_size=1, strategic_metadata=strategic_metadata)
+                producer_client.insert_submission(
+                    [1], chunk_size=1, strategic_metadata=strategic_metadata
+                )
             )
         with pytest.raises(TooManyMatchingSubmissionsError) as exc:
             assert len(inserted) == max_ + 1
-            producer_client.lookup_submission_ids_by_strategic_metadata(strategic_metadata)
+            producer_client.lookup_submission_ids_by_strategic_metadata(
+                strategic_metadata
+            )
         assert exc.type is TooManyMatchingSubmissionsError
         assert exc.value.max_submissions == max_
diff --git a/opsqueue/src/common/submission.rs b/opsqueue/src/common/submission.rs
index c71763c6..b2cd4166 100644
--- a/opsqueue/src/common/submission.rs
+++ b/opsqueue/src/common/submission.rs
@@ -585,14 +585,11 @@ pub mod db {
         let mut query_builder: QueryBuilder<Sqlite> =
             QueryBuilder::new("SELECT id FROM submissions");
         // Inner join for each piece of strategic metadata.
-        for i in 0..strategic_metadata.len() {
+        for (i, (key, value)) in strategic_metadata.iter().enumerate() {
             query_builder.push(format!(
-                " JOIN submissions_metadata AS s{i} ON s{i}.submission_id = submissions.id"
+                " INNER JOIN submissions_metadata AS s{i} ON s{i}.submission_id = submissions.id"
             ));
-        }
-        for (i, (key, value)) in strategic_metadata.iter().enumerate() {
-            query_builder.push(if i == 0 { " WHERE" } else { " AND" });
-            query_builder.push(format!(" s{i}.metadata_key = "));
+            query_builder.push(format!(" AND s{i}.metadata_key = "));
             query_builder.push_bind(key.clone());
             query_builder.push(format!(" AND s{i}.metadata_value = "));
             query_builder.push_bind(*value);
@@ -1143,11 +1140,10 @@ pub mod test {
           id
         FROM
           submissions
-          JOIN submissions_metadata AS s0 ON s0.submission_id = submissions.id
-          JOIN submissions_metadata AS s1 ON s1.submission_id = submissions.id
-        WHERE
-          s0.metadata_key = ?
+          INNER JOIN submissions_metadata AS s0 ON s0.submission_id = submissions.id
+          AND s0.metadata_key = ?
           AND s0.metadata_value = ?
+          INNER JOIN submissions_metadata AS s1 ON s1.submission_id = submissions.id
           AND s1.metadata_key = ?
           AND s1.metadata_value = ?
         ORDER BY

From 707485426084d79206e26a073163d92f4e14cc8d Mon Sep 17 00:00:00 2001
From: "jeremy.barisch.rooney@channable.com"
 <jeremy.barisch.rooney@channable.com>
Date: Wed, 15 Jul 2026 14:00:38 +0200
Subject: [PATCH 16/16] fixup! Add
 ProducerClient.lookup_submission_ids_by_strategic_metadata

bump version to 0.36.0
---
 Cargo.lock | 4 ++--
 Cargo.toml | 2 +-
 2 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 30e6cc6a..6a02a4a0 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2270,7 +2270,7 @@ dependencies = [
 
 [[package]]
 name = "opsqueue"
-version = "0.35.3"
+version = "0.36.3"
 dependencies = [
  "anyhow",
  "arc-swap",
@@ -2325,7 +2325,7 @@ dependencies = [
 
 [[package]]
 name = "opsqueue_python"
-version = "0.35.3"
+version = "0.36.3"
 dependencies = [
  "anyhow",
  "chrono",
diff --git a/Cargo.toml b/Cargo.toml
index 736d84ab..76d0baf6 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -9,7 +9,7 @@ members = [
 ]
 
 [workspace.package]
-version = "0.35.3"
+version = "0.36.3"
 
 [workspace.lints.clippy]
 cargo =  { level = "warn", priority = -1 }