Lookup submision IDs by strategic metadata by jerbaroo · Pull Request #112 · channable/opsqueue

jerbaroo · 2026-06-08T15:45:15Z

This PR adds the lookup_submission_ids_by_strategic_metadata method to ProducerClient, which will return the submission IDs of in-progress subsmissions that match ALL of the provided strategic metadata key-value pairs.

    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.

        """

ReinierMaas · 2026-06-23T14:21:08Z

Open question shouldn't an empty StrategicMetadataMap return all submissions instead of raising an error on that?
Also this is an easy way to DDoS OpsQueue if the number of returned submissions is large.

jerbaroo · 2026-06-23T14:49:19Z

Open question shouldn't an empty StrategicMetadataMap return all submissions instead of raising an error on that?

It depends on the API contract we want to provide. An argument in favour of throwing an error is that if a caller is using an API to "search for submissions matching some strategic metadata" but they haven't provided any strategic metadata, then they possibly have an issue in their code that they should be alerted to. Also, the SQLite query will need a special case to handle this, more code for something we don't have evidence for that it is a usecase we want to support.

ReinierMaas

I have a few comment and would like to point out that for the other queries we also assert the query plan. This allows us to spot potentially very bad execution behaviour.

ReinierMaas · 2026-06-23T14:24:28Z

+        Raises:
+        - `LookupIdsWithEmptyStrategicMetadataError` if the provided
+          'strategic_metadata' contained no key-value pairs to look for.


Can this potentially also raise:

- `InternalProducerClientError` if there is a low-level internal error.

Yes, good spot. The rust producer/client.rs code lifts network and decoding errors into this type. Documented in b557df0

ReinierMaas · 2026-06-23T14:26:40Z

+    def lookup_submission_ids_by_strategic_metadata(
+        self, strategic_metadata: dict[str, int]
+    ) -> list[SubmissionId]:


I checked the other APIs on the producer and they either provide a single element or an iterator so that it can be lazily evaluated and doesn't need to be materialized in memory all at once.

The other ProducerClient APIs which are returning iterators are returning chunk iterators. These chunks are deterministically-addressable. Internally, the iterator is keeping track of a submission prefix, a current index, and a max index, then on each call to next() the iterator is performing a network request (GET <object-storage-path><prefix>/<current-index>-out.bin) to fetch the chunk.

This new API is different (not chunks, not querying object storage) from the existing APIs that return iterators, so would need some new wiring put in place to support streaming. But since we are only dealing with submission IDs, it's less of a memory concern. A pragmatic solution for now might be to just implement the configurable upper bound + add some metrics and keep an eye on it.

ReinierMaas · 2026-06-23T14:42:59Z

+async fn lookup_submission_ids_by_strategic_metadata(
+    State(state): State<ServerState>,
+    extract::Json(strategic_metadata): extract::Json<StrategicMetadataMap>,
+) -> Result<Json<Vec<SubmissionId>>, ServerError> {


Our SubmissionIds are close to u64::max in serialized size, which is reasonable for our random (snowflake) identifiers. This issues the following serialized to JSON response size:

1 ([) + elements * 20 + elements (,|]) characters

For ~1 million this is ~20MB. We can put that as an configurable upper bound to return to the client with an explicit error if it goes over that so that people can update the settings and return the larger response knowingly.

Configurable upper bound added in 54bc50b with a roundtrip test in be5f6d2

It might be a bit over-engineered with a newtype that ensures we don't overflow when adding (+1) when building the SQLite query, but.. types good right ? 😅

ReinierMaas · 2026-06-23T14:46:49Z

+        Raises:
+        - `LookupIdsWithEmptyStrategicMetadataError` if the provided
+          'strategic_metadata' contained no key-value pairs to look for.


I would change the interface to raise an explicit error if there are too many matching Submissions, we need that for DoS protection anyway, but allow an empty strategic_metadata set to be processed.

SGTM. Added support for empty strategic_metadata in 8057405. Will add the streaming / DoS protection in separate commits.

Document InternalProducerClientError raised

Support empty strategic metadata

Add configurable limit to submissions lookup

…etadata Test it

jerbaroo · 2026-07-09T14:27:18Z

I have a few comment and would like to point out that for the other queries we also assert the query plan. This allows us to spot potentially very bad execution behaviour.

Query plan asserted in c6c9f04

assert query plan

ReinierMaas

I have a few code simplifications and one API generalisation.

ReinierMaas · 2026-07-10T07:48:43Z

I am not sure these timeouts are long enough on CI.

I had the same experience locally and made it configurable for local use:

https://github.com/channable/opsqueue/pull/122/changes#diff-deb9bb56fb122db0b605aa5b63f95a4665c905b18dd670e1fa6c877576a94ff1

ReinierMaas · 2026-07-10T07:52:25Z


+impl From<CError<TooManyMatchingSubmissions>> for PyErr {
+    fn from(value: CError<TooManyMatchingSubmissions>) -> Self {
+        TooManyMatchingSubmissionsError::new_err(value.0 .0)


Suggested change

TooManyMatchingSubmissionsError::new_err(value.0 .0)

TooManyMatchingSubmissionsError::new_err(value.0.0)

ReinierMaas · 2026-07-10T08:06:58Z

+class LookupIdsWithEmptyStrategicMetadataError(Exception):
+    pass


Is this error still produced?

ReinierMaas · 2026-07-10T08:12:08Z

 pub type StrategicMetadataMap = FxHashMap<String, MetaStateVal>;
+
+/// Maximum number of submissions a lookup may return.
+/// Constructor checks: MaxSubmissions + 1 <= i64::MAX;


We should also prevent 0 from appearing, i.e. use NonZeroU64 to store the value.

ReinierMaas · 2026-07-10T08:20:29Z

+        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})


We can also test the endpoint keeps functioning until the one that brings it over the edge, this prevents off-by-one errors:

Suggested change

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})

inserted = []

for _ in range(max_ + 1):

assert producer_client.lookup_submission_ids_by_strategic_metadata({"k": 1}) == inserted

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.

producer_client.lookup_submission_ids_by_strategic_metadata({"k": 1})

ReinierMaas · 2026-07-10T08:21:30Z

+        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
+          ?
+        ");


This idea turned out to be so beautiful! It makes it easy to review the generated queries.

ReinierMaas · 2026-07-10T08:25:42Z

+                " 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");


This is to scope it to the InProgress submissions. We could make this INNER JOIN table configurable based on the requested SubmissionState enum.

ReinierMaas · 2026-07-10T09:32:42Z

+        // 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!(
+                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);
+        }
+        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(),
+        )


We can simplify this a bit further to make it easier to follow:

Suggested change

// 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!(

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);

}

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(),

)

let ids = if strategic_metadata.is_empty() {

// 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.

query_scalar!(

r#"SELECT id AS "id: SubmissionId" FROM submissions ORDER BY id LIMIT ?"#,

limit

)

.fetch_all(conn.get_inner())

.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) {

// If 'max_submissions' results is exceeded return an explicit error.

Err(E::R(TooManyMatchingSubmissions(u64::from(max_submissions))))

} else {

Ok(ids)

}

Note that if we add a SubmissionState selector we need to switch out the FROM submissions with the correct submission(_*)? tables.

ReinierMaas · 2026-07-10T09:34:48Z

+        limit: i64,
+    ) -> QueryBuilder<'_, Sqlite> {
+        let mut query_builder: QueryBuilder<Sqlite> =
+            QueryBuilder::new("SELECT s0.submission_id FROM submissions_metadata as s0");


In the other query you added AS "id: SubmissionId" don't we need that here?

jerbaroo requested a review from Qqwy June 8, 2026 15:58

jerbaroo force-pushed the jerbaroo-64-lookup-by-strategic-metadata branch from 3b0ca52 to 6f9e1e3 Compare June 8, 2026 15:58

jerbaroo linked an issue Jun 8, 2026 that may be closed by this pull request

Support checking whether a submission currently exists in the queue for a particular piece of strategic metadata #64

Open

jerbaroo requested a review from ReinierMaas June 23, 2026 12:30

ReinierMaas reviewed Jun 23, 2026

View reviewed changes

jerbaroo force-pushed the jerbaroo-64-lookup-by-strategic-metadata branch 3 times, most recently from 8057405 to 736a7fd Compare July 7, 2026 15:48

jerbaroo added 4 commits July 7, 2026 17:48

Add ProducerClient.lookup_submission_ids_by_strategic_metadata

ddee243

fixup! Add ProducerClient.lookup_submission_ids_by_strategic_metadata

36606c1

Document InternalProducerClientError raised

fixup! Add ProducerClient.lookup_submission_ids_by_strategic_metadata

f329863

Support empty strategic metadata

fixup! Add ProducerClient.lookup_submission_ids_by_strategic_metadata

54bc50b

Add configurable limit to submissions lookup

jerbaroo force-pushed the jerbaroo-64-lookup-by-strategic-metadata branch from 736a7fd to 54bc50b Compare July 7, 2026 15:48

fixup! fixup! Add ProducerClient.lookup_submission_ids_by_strategic_m…

be5f6d2

…etadata Test it

ReinierMaas assigned jerbaroo Jul 9, 2026

jerbaroo force-pushed the jerbaroo-64-lookup-by-strategic-metadata branch from 74a4aa8 to 841d1a3 Compare July 9, 2026 14:26

jerbaroo requested a review from ReinierMaas July 9, 2026 14:27

jerbaroo force-pushed the jerbaroo-64-lookup-by-strategic-metadata branch from 841d1a3 to c6c9f04 Compare July 9, 2026 15:57

fixup! Add ProducerClient.lookup_submission_ids_by_strategic_metadata

6c0256e

assert query plan

jerbaroo force-pushed the jerbaroo-64-lookup-by-strategic-metadata branch from c6c9f04 to 6c0256e Compare July 9, 2026 16:31

ReinierMaas reviewed Jul 10, 2026

View reviewed changes

	TooManyMatchingSubmissionsError::new_err(value.0 .0)
	TooManyMatchingSubmissionsError::new_err(value.0.0)

		class LookupIdsWithEmptyStrategicMetadataError(Exception):
		pass

Uh oh!

Conversation

jerbaroo commented Jun 8, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ReinierMaas commented Jun 23, 2026

Uh oh!

jerbaroo commented Jun 23, 2026

Uh oh!

ReinierMaas left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jerbaroo Jul 7, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jerbaroo Jul 7, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jerbaroo commented Jul 9, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ReinierMaas left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

jerbaroo commented Jun 8, 2026 •

edited

Loading

jerbaroo Jul 7, 2026 •

edited

Loading

jerbaroo Jul 7, 2026 •

edited

Loading

jerbaroo commented Jul 9, 2026 •

edited

Loading