Add Jobmachine delegation

libs/opsqueue_python/python/opsqueue/producer.py

@@ -94,12 +94,16 @@ def run_submission(
         serialization_format: SerializationFormat = DEFAULT_SERIALIZATION_FORMAT,
         metadata: None | bytes = None,
         strategic_metadata: None | dict[str, int] = None,
+        paused: bool = False,
     ) -> Iterator[Any]:
         """
         Inserts a submission into the queue, and blocks until it is completed.
 
         Chunking is done automatically, based on the provided chunk size.
 
+        If `paused` is True, the submission will not be dispatched to consumers until
+        it is explicitly unpaused (e.g. via the delegation `job_delegate` endpoint).
+
         If the submission fails, an exception will be raised.
         (If opsqueue or the object storage cannot be reached, exceptions will also be raised).
 
@@ -114,6 +118,7 @@ def run_submission(
                 metadata=metadata,
                 strategic_metadata=strategic_metadata,
                 chunk_size=chunk_size,
+                paused=paused,
             )
             return _unchunk_iterator(results_iter, serialization_format)
 
@@ -125,6 +130,7 @@ async def async_run_submission(
         serialization_format: SerializationFormat = DEFAULT_SERIALIZATION_FORMAT,
         metadata: None | bytes = None,
         strategic_metadata: None | dict[str, int] = None,
+        paused: bool = False,
     ) -> AsyncIterator[Any]:
         tracer = trace.get_tracer("opsqueue.producer")
         with tracer.start_as_current_span("run_submission"):
@@ -133,6 +139,7 @@ async def async_run_submission(
                 metadata=metadata,
                 strategic_metadata=strategic_metadata,
                 chunk_size=chunk_size,
+                paused=paused,
             )
             return _async_unchunk_iterator(results_iter, serialization_format)
 
@@ -144,13 +151,17 @@ def insert_submission(
         serialization_format: SerializationFormat = DEFAULT_SERIALIZATION_FORMAT,
         metadata: None | bytes = None,
         strategic_metadata: None | dict[str, int] = None,
+        paused: bool = False,
     ) -> SubmissionId:
         """
         Inserts a submission into the queue,
         returning an ID you can use to track the submission's progress afterwards.
 
         Chunking is done automatically, based on the provided chunk size.
 
+        If `paused` is True, the submission will not be dispatched to consumers until
+        it is explicitly unpaused (e.g. via the delegation `job_delegate` endpoint).
+
         Raises:
         - `ChunkSizeIsZeroError` if passing an incorrect chunk size of zero;
         - `InternalProducerClientError` if there is a low-level internal error.
@@ -160,6 +171,7 @@ def insert_submission(
             metadata=metadata,
             strategic_metadata=strategic_metadata,
             chunk_size=chunk_size,
+            paused=paused,
         )
 
     def blocking_stream_completed_submission(
@@ -209,10 +221,14 @@ def run_submission_chunks(
         metadata: None | bytes = None,
         strategic_metadata: None | dict[str, int] = None,
         chunk_size: None | int = None,
+        paused: bool = False,
     ) -> Iterator[bytes]:
         """
         Inserts an already-chunked submission into the queue, and blocks until it is completed.
 
+        If `paused` is True, the submission will not be dispatched to consumers until
+        it is explicitly unpaused (e.g. via the delegation `job_delegate` endpoint).
+
         If the submission fails, an exception will be raised.
         (If opsqueue or the object storage cannot be reached, exceptions will also be raised).
 
@@ -226,6 +242,7 @@ def run_submission_chunks(
             metadata=metadata,
             strategic_metadata=strategic_metadata,
             chunk_size=chunk_size,
+            paused=paused,
         )
         return self.blocking_stream_completed_submission_chunks(submission_id)
 
@@ -236,6 +253,7 @@ async def async_run_submission_chunks(
         metadata: None | bytes = None,
         strategic_metadata: None | dict[str, int] = None,
         chunk_size: None | int = None,
+        paused: bool = False,
     ) -> AsyncIterator[bytes]:
         # NOTE: the insertion is not currently async.
         # Why? Simplicity. This is unlikely to be the bottleneck
@@ -246,6 +264,7 @@ async def async_run_submission_chunks(
             metadata=metadata,
             strategic_metadata=strategic_metadata,
             chunk_size=chunk_size,
+            paused=paused,
         )
 
         return await self.async_stream_completed_submission_chunks(submission_id)
@@ -257,11 +276,15 @@ def insert_submission_chunks(
         metadata: None | bytes = None,
         strategic_metadata: None | dict[str, int] = None,
         chunk_size: None | int = None,
+        paused: bool = False,
     ) -> SubmissionId:
         """
         Inserts an already-chunked submission into the queue,
         returning an ID you can use to track the submission's progress afterwards.
 
+        If `paused` is True, the submission will not be dispatched to consumers until
+        it is explicitly unpaused (e.g. via the delegation `job_delegate` endpoint).
+
         Raises:
         - `InternalProducerClientError` if there is a low-level internal error.
         """
@@ -272,6 +295,7 @@ def insert_submission_chunks(
             metadata=metadata,
             strategic_metadata=strategic_metadata,
             chunk_size=chunk_size,
+            paused=paused,
             otel_trace_carrier=otel_trace_carrier,
         )
 

libs/opsqueue_python/src/producer.rs

@@ -192,13 +192,14 @@ impl ProducerClient {
     /// Directly inserts a submission without sending the chunks to GCS
     /// (but immediately embedding them in the DB).
     /// NOTE: This does not support StrategicMetadata currently
-    #[pyo3(signature = (chunk_contents, metadata=None, chunk_size=None, otel_trace_carrier=CarrierMap::default()))]
+    #[pyo3(signature = (chunk_contents, metadata=None, chunk_size=None, paused=false, otel_trace_carrier=CarrierMap::default()))]
     pub fn insert_submission_direct(
         &self,
         py: Python<'_>,
         chunk_contents: Vec<chunk::Content>,
         metadata: Option<submission::Metadata>,
         chunk_size: Option<u64>,
+        paused: bool,
         otel_trace_carrier: CarrierMap,
     ) -> CPyResult<SubmissionId, E<FatalPythonException, InternalProducerClientError>> {
         let strategic_metadata = Default::default();
@@ -211,6 +212,7 @@ impl ProducerClient {
                 },
                 metadata,
                 strategic_metadata,
+                paused,
             };
             self.block_unless_interrupted(async move {
                 self.producer_client
@@ -222,7 +224,7 @@ impl ProducerClient {
         })
     }
 
-    #[pyo3(signature = (chunk_contents, metadata=None, strategic_metadata=None, chunk_size=None, otel_trace_carrier=CarrierMap::default()))]
+    #[pyo3(signature = (chunk_contents, metadata=None, strategic_metadata=None, chunk_size=None, paused=false, otel_trace_carrier=CarrierMap::default()))]
     #[allow(clippy::type_complexity)]
     pub fn insert_submission_chunks(
         &self,
@@ -231,6 +233,7 @@ impl ProducerClient {
         metadata: Option<submission::Metadata>,
         strategic_metadata: Option<StrategicMetadataMap>,
         chunk_size: Option<i64>,
+        paused: bool,
         otel_trace_carrier: CarrierMap,
     ) -> CPyResult<
         SubmissionId,
@@ -271,6 +274,7 @@ impl ProducerClient {
                     },
                     metadata,
                     strategic_metadata: strategic_metadata.unwrap_or_default(),
+                    paused,
                 };
                 self.producer_client
                     .insert_submission(&submission, &otel_trace_carrier)
@@ -312,7 +316,7 @@ impl ProducerClient {
         })
     }
 
-    #[pyo3(signature = (chunk_contents, metadata=None, strategic_metadata=None, chunk_size=None, otel_trace_carrier=CarrierMap::default()))]
+    #[pyo3(signature = (chunk_contents, metadata=None, strategic_metadata=None, chunk_size=None, paused=false, otel_trace_carrier=CarrierMap::default()))]
     #[allow(clippy::type_complexity)]
     pub fn run_submission_chunks(
         &self,
@@ -321,6 +325,7 @@ impl ProducerClient {
         metadata: Option<submission::Metadata>,
         strategic_metadata: Option<StrategicMetadataMap>,
         chunk_size: Option<i64>,
+        paused: bool,
         otel_trace_carrier: CarrierMap,
     ) -> CPyResult<
         PyChunksIter,
@@ -338,6 +343,7 @@ impl ProducerClient {
                 metadata,
                 strategic_metadata,
                 chunk_size,
+                paused,
                 otel_trace_carrier,
             )
             .map_err(|CError(e)| {

opsqueue/Cargo.toml

@@ -102,7 +102,8 @@ server-logic = [
     "dep:tower-http",
     "dep:axum-prometheus",
     "dep:sentry",
-    "dep:sentry-tracing"
+    "dep:sentry-tracing",
+    "dep:reqwest",
     ]
 # Dependencies only in use by the client libraries:
 client-logic = [

opsqueue/migrations/20260709130000_add_jm_task_id.down.sql

@@ -0,0 +1,4 @@
+ALTER TABLE submissions DROP COLUMN jm_task_id;
+ALTER TABLE submissions_completed DROP COLUMN jm_task_id;
+ALTER TABLE submissions_failed DROP COLUMN jm_task_id;
+ALTER TABLE submissions_cancelled DROP COLUMN jm_task_id;

opsqueue/migrations/20260709130000_add_jm_task_id.up.sql

@@ -0,0 +1,4 @@
+ALTER TABLE submissions ADD COLUMN jm_task_id TEXT;
+ALTER TABLE submissions_completed ADD COLUMN jm_task_id TEXT;
+ALTER TABLE submissions_failed ADD COLUMN jm_task_id TEXT;
+ALTER TABLE submissions_cancelled ADD COLUMN jm_task_id TEXT;

opsqueue/migrations/20260709160000_jm_task_id_indexes.down.sql

@@ -0,0 +1,4 @@
+DROP INDEX IF EXISTS submissions_by_jm_task_id;
+DROP INDEX IF EXISTS submissions_completed_by_jm_task_id;
+DROP INDEX IF EXISTS submissions_failed_by_jm_task_id;
+DROP INDEX IF EXISTS submissions_cancelled_by_jm_task_id;

opsqueue/migrations/20260709160000_jm_task_id_indexes.up.sql

@@ -0,0 +1,4 @@
+CREATE INDEX submissions_by_jm_task_id ON submissions (jm_task_id) WHERE jm_task_id IS NOT NULL;
+CREATE INDEX submissions_completed_by_jm_task_id ON submissions_completed (jm_task_id) WHERE jm_task_id IS NOT NULL;
+CREATE INDEX submissions_failed_by_jm_task_id ON submissions_failed (jm_task_id) WHERE jm_task_id IS NOT NULL;
+CREATE INDEX submissions_cancelled_by_jm_task_id ON submissions_cancelled (jm_task_id) WHERE jm_task_id IS NOT NULL;

opsqueue/migrations/20260709170000_add_paused_to_submissions.down.sql

@@ -0,0 +1 @@
+ALTER TABLE submissions DROP COLUMN paused;

opsqueue/migrations/20260709170000_add_paused_to_submissions.up.sql

@@ -0,0 +1 @@
+ALTER TABLE submissions ADD COLUMN paused INTEGER NOT NULL DEFAULT 0;

opsqueue/src/common/chunk.rs

@@ -779,6 +779,7 @@ pub mod test {
             None,
             StrategicMetadataMap::default(),
             ChunkSize::default(),
+            false,
             &mut conn,
         )
         .await