Sort submissions by metadata count

Cargo.toml

@@ -35,6 +35,7 @@ http = { version = "1.4.0" }
 humantime = { version = "2.1.0" }
 insta = { version = "1.47.2" }
 itertools = { version = "0.15.0" }
+libsqlite3-sys = { version = "0.30.1" }
 moka = { version = "0.12.15", features = ["sync"] }
 moro-local = { version = "0.4.0" }
 object_store = { version = "0.14.0", features = ["gcp", "http"] }

libs/opsqueue_python/tests/test_roundtrip.py

@@ -667,3 +667,32 @@ def test_lookup_too_many_submission_ids_by_strategic_metadata() -> None:
             )
         assert exc.type is TooManyMatchingSubmissionsError
         assert exc.value.max_submissions == max_
+
+
+def test_prefer_distinct_strategy_fairness(opsqueue: OpsqueueProcess) -> None:
+    """Test the PreferDistinct strategy fairly interleaves chunks from different
+    submissions based on their strategic metadata.
+
+    """
+    url = "file:///tmp/opsqueue/test_prefer_distinct_fairness"
+    producer_client = ProducerClient(f"localhost:{opsqueue.port}", url)
+    consumer_client = ConsumerClient(f"localhost:{opsqueue.port}", url)
+    company_ids = [1, 2, 3]
+    chunks_per_company = 4
+    company_id_per_submission = {}
+    for company_id in company_ids:
+        sub_id = producer_client.insert_submission(
+            [company_id] * chunks_per_company,
+            chunk_size=1,
+            strategic_metadata={"company_id": company_id},
+        )
+        company_id_per_submission[sub_id] = company_id
+    strategy = strategy_from_description(("PreferDistinct", "company_id", "Oldest"))
+    reserved_company_order = []
+    # Fetch 1 chunk at a time. Because we don't complete chunks, opsqueue's
+    # metastate tracks them as reserved, increasing the busy count for that
+    # company.
+    for _ in range(len(company_ids) * chunks_per_company):
+        [chunk] = consumer_client.reserve_chunks(strategy=strategy)
+        reserved_company_order.append(company_id_per_submission[chunk.submission_id])
+    assert reserved_company_order == [1, 2, 3] * chunks_per_company

opsqueue/Cargo.toml

@@ -29,6 +29,7 @@ ux.workspace = true
 anyhow.workspace = true
 # Database:
 sqlx = { workspace = true, optional = true }
+libsqlite3-sys = { workspace = true, optional = true }
 # Serialization:
 serde.workspace = true
 serde_json.workspace = true
@@ -69,7 +70,6 @@ clap.workspace = true
 humantime.workspace = true
 
 dashmap.workspace = true
-crossbeam-skiplist.workspace = true
 sqlformat.workspace = true
 workspace-hack.workspace = true
 
@@ -90,6 +90,7 @@ required-features = ["server-logic"]
 # Dependencies only in use by the server-logic:
 server-logic = [
     "dep:sqlx",
+    "dep:libsqlite3-sys",
     "dep:opentelemetry-otlp",
     "dep:opentelemetry-semantic-conventions",
     "dep:moka",