diff --git a/Cargo.lock b/Cargo.lock index 9d5960f..fe470f8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -595,16 +595,6 @@ dependencies = [ "crossbeam-utils", ] -[[package]] -name = "crossbeam-skiplist" -version = "0.1.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "df29de440c58ca2cc6e587ec3d22347551a32435fbde9d2bff64e78a9ffa151b" -dependencies = [ - "crossbeam-epoch", - "crossbeam-utils", -] - [[package]] name = "crossbeam-utils" version = "0.8.22" @@ -2111,7 +2101,6 @@ dependencies = [ "chrono", "ciborium", "clap", - "crossbeam-skiplist", "dashmap", "either", "futures", @@ -2119,6 +2108,7 @@ dependencies = [ "humantime", "insta", "itertools 0.15.0", + "libsqlite3-sys", "moka", "moro-local", "object_store", @@ -3233,7 +3223,6 @@ dependencies = [ "log", "memchr", "percent-encoding", - "rustls", "serde", "serde_json", "sha2 0.10.9", @@ -3243,7 +3232,6 @@ dependencies = [ "tokio-stream", "tracing", "url", - "webpki-roots", ] [[package]] @@ -4360,12 +4348,10 @@ checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e" name = "workspace-hack" version = "0.1.0" dependencies = [ - "aws-lc-rs", "base64", "bitflags", "cc", "chrono", - "crossbeam-epoch", "crossbeam-utils", "either", "event-listener", diff --git a/Cargo.toml b/Cargo.toml index 68ec307..44ce405 100644 --- a/Cargo.toml +++ b/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"] } diff --git a/libs/opsqueue_python/tests/test_roundtrip.py b/libs/opsqueue_python/tests/test_roundtrip.py index 23d9a14..2e887ac 100644 --- a/libs/opsqueue_python/tests/test_roundtrip.py +++ b/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 diff --git a/opsqueue/Cargo.toml b/opsqueue/Cargo.toml index 82d562a..75daf32 100644 --- a/opsqueue/Cargo.toml +++ b/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", diff --git a/opsqueue/benches/chunks_select_bench.svg b/opsqueue/benches/chunks_select_bench.svg index 1df4bd6..fdcebee 100644 --- a/opsqueue/benches/chunks_select_bench.svg +++ b/opsqueue/benches/chunks_select_bench.svg @@ -39,84 +39,84 @@ z - - - + + - - - + + @@ -124,16 +124,16 @@ z +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - - + @@ -210,11 +210,11 @@ z +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -264,11 +264,11 @@ z +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -305,11 +305,11 @@ z +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -352,11 +352,11 @@ z +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -404,16 +404,16 @@ z +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - - + @@ -421,11 +421,11 @@ L 0 2 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -433,11 +433,11 @@ L 52.720915 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -445,11 +445,11 @@ L 57.915728 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -457,11 +457,11 @@ L 93.13386 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -469,11 +469,11 @@ L 111.016873 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -481,11 +481,11 @@ L 123.705072 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -493,11 +493,11 @@ L 133.546804 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -505,11 +505,11 @@ L 141.588085 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -517,11 +517,11 @@ L 148.386891 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -529,11 +529,11 @@ L 154.276285 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -541,11 +541,11 @@ L 159.471098 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -553,11 +553,11 @@ L 194.689229 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -565,11 +565,11 @@ L 212.572242 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -577,11 +577,11 @@ L 225.260442 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -589,11 +589,11 @@ L 235.102174 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -601,11 +601,11 @@ L 243.143455 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -613,11 +613,11 @@ L 249.942261 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -625,11 +625,11 @@ L 255.831654 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -637,11 +637,11 @@ L 261.026468 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -649,11 +649,11 @@ L 296.244599 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -661,11 +661,11 @@ L 314.127612 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -673,11 +673,11 @@ L 326.815811 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -685,11 +685,11 @@ L 336.657543 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -697,11 +697,11 @@ L 344.698824 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -709,11 +709,11 @@ L 351.49763 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -721,11 +721,11 @@ L 357.387024 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -733,11 +733,11 @@ L 362.581837 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -745,11 +745,11 @@ L 397.799968 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -757,11 +757,11 @@ L 415.682981 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -769,11 +769,11 @@ L 428.371181 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -781,11 +781,11 @@ L 438.212913 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -793,11 +793,11 @@ L 446.254194 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -805,11 +805,11 @@ L 453.053 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -817,11 +817,11 @@ L 458.942393 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -829,11 +829,11 @@ L 464.137207 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -841,11 +841,11 @@ L 499.355338 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -853,11 +853,11 @@ L 517.238351 48.482812 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -1251,23 +1251,23 @@ z - + - - + - + @@ -1276,18 +1276,18 @@ L -3.5 0 - + - + - + @@ -1296,18 +1296,18 @@ L 532.078437 270.441087 - + - + - + @@ -1316,18 +1316,18 @@ L 532.078437 214.951518 - + - + - + @@ -1336,18 +1336,18 @@ L 532.078437 159.46195 - + - + - + @@ -1358,11 +1358,11 @@ L 532.078437 103.972381 +" clip-path="url(#p9766e11a9e)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + @@ -1388,558 +1388,486 @@ z - + - - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + @@ -2218,25 +2146,25 @@ z - - + + - - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + + - - + + - - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + + @@ -2516,13 +2444,13 @@ Q 319.634844 82.684219 321.434844 82.684219 z " style="fill: #ffffff; opacity: 0.8; stroke: #cccccc; stroke-linejoin: miter"/> - + - + @@ -2657,13 +2585,13 @@ z - + - + @@ -2720,96 +2648,96 @@ z - - - + + - - - + + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + @@ -2822,14 +2750,14 @@ L 598.962647 48.482812 - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + @@ -2842,14 +2770,14 @@ L 700.518017 48.482812 - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + @@ -2862,14 +2790,14 @@ L 802.073386 48.482812 - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + @@ -2882,14 +2810,14 @@ L 903.628756 48.482812 - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + @@ -2902,458 +2830,458 @@ L 1005.184125 48.482812 - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + @@ -3391,115 +3319,115 @@ L 1066.32655 48.482812 - - - + + + - + - + - + - - - + + + - + - + - + - - - + + + - + - + - + - - - + + + - + - + - + - - - + + + - + - + - + - - + + +" clip-path="url(#pba62c91b52)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + @@ -3511,543 +3439,507 @@ L 1068.478437 48.482812 - - - + + + - + - + - - - + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + @@ -4109,74 +4001,74 @@ L 1068.478437 51.064955 - - - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + + + + - - - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + + + + @@ -4242,13 +4134,13 @@ Q 856.034844 82.684219 857.834844 82.684219 z " style="fill: #ffffff; opacity: 0.8; stroke: #cccccc; stroke-linejoin: miter"/> - + - + @@ -4294,13 +4186,13 @@ L 877.634844 60.271406 - + - + @@ -4327,128 +4219,128 @@ z - - - + + - - - + + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + @@ -4461,14 +4353,14 @@ L 1141.75686 48.482812 - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + @@ -4481,14 +4373,14 @@ L 1241.929174 48.482812 - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + @@ -4501,14 +4393,14 @@ L 1342.101487 48.482812 - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + @@ -4521,14 +4413,14 @@ L 1442.273801 48.482812 - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + @@ -4541,470 +4433,470 @@ L 1542.446114 48.482812 - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + - + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + @@ -5042,115 +4934,115 @@ L 1602.755856 48.482812 - - - + + + - + - + - + - - - + + + - + - + - + - - - + + + - + - + - + - - - + + + - + - + - + - - - + + + - + - + - + - - + + +" clip-path="url(#p05342ab53a)" style="fill: none; stroke-dasharray: 0.8,1.32; stroke-dashoffset: 0; stroke: #b0b0b0; stroke-opacity: 0.5; stroke-width: 0.8"/> - + - + @@ -5162,567 +5054,507 @@ L 1604.878437 48.482812 + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + @@ -5784,106 +5616,106 @@ L 1604.878437 50.987189 - - - - - - - - - - - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + + + + + + + + + + + + - - - - - - - - - - - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + + + + + + + + + + + + @@ -5934,13 +5766,13 @@ Q 1392.434844 82.684219 1394.234844 82.684219 z " style="fill: #ffffff; opacity: 0.8; stroke: #cccccc; stroke-linejoin: miter"/> - + - + @@ -5986,13 +5818,13 @@ L 1414.034844 60.271406 - + - + @@ -6084,13 +5916,13 @@ z - + - + - + diff --git a/opsqueue/migrations/20260803133844_add_random_order_index_to_submissions.down.sql b/opsqueue/migrations/20260803133844_add_random_order_index_to_submissions.down.sql new file mode 100644 index 0000000..6432542 --- /dev/null +++ b/opsqueue/migrations/20260803133844_add_random_order_index_to_submissions.down.sql @@ -0,0 +1,2 @@ +DROP INDEX random_submissions_order; +ALTER TABLE submissions DROP COLUMN random_order; diff --git a/opsqueue/migrations/20260803133844_add_random_order_index_to_submissions.up.sql b/opsqueue/migrations/20260803133844_add_random_order_index_to_submissions.up.sql new file mode 100644 index 0000000..ae5ee78 --- /dev/null +++ b/opsqueue/migrations/20260803133844_add_random_order_index_to_submissions.up.sql @@ -0,0 +1,9 @@ +-- Uses the same formula as '20250803174028_better_random_order_formula.up.sql'. +ALTER TABLE submissions ADD COLUMN random_order INTEGER NOT NULL GENERATED ALWAYS AS ( + (((id + (id >> 22)) % 65536) * 40503) % 65536 +) VIRTUAL; + +CREATE INDEX random_submissions_order ON submissions ( + random_order + , id +); diff --git a/opsqueue/opsqueue_example_database_schema.db b/opsqueue/opsqueue_example_database_schema.db index 83e0994..e13b54b 100644 Binary files a/opsqueue/opsqueue_example_database_schema.db and b/opsqueue/opsqueue_example_database_schema.db differ diff --git a/opsqueue/src/common/chunk.rs b/opsqueue/src/common/chunk.rs index c27c17e..9010b92 100644 --- a/opsqueue/src/common/chunk.rs +++ b/opsqueue/src/common/chunk.rs @@ -130,6 +130,18 @@ impl From for i64 { } } +impl TryFrom for ChunkIndex { + type Error = crate::common::errors::TryFromIntError; + + fn try_from(value: i64) -> Result { + if value < 0 { + return Err(crate::common::errors::TryFromIntError(())); + } + + Ok(Self(u63::new(value.cast_unsigned()))) + } +} + impl TryFrom for ChunkIndex { type Error = crate::common::errors::TryFromIntError; fn try_from(value: u64) -> Result { diff --git a/opsqueue/src/common/submission.rs b/opsqueue/src/common/submission.rs index 991744d..0b6091a 100644 --- a/opsqueue/src/common/submission.rs +++ b/opsqueue/src/common/submission.rs @@ -580,12 +580,10 @@ pub mod db { r#" SELECT id AS "id: SubmissionId" FROM submissions WHERE prefix = $1 UNION ALL - SELECT id AS "id: SubmissionId" FROM submissions_completed WHERE prefix = $2 + SELECT id AS "id: SubmissionId" FROM submissions_completed WHERE prefix = $1 UNION ALL - SELECT id AS "id: SubmissionId" FROM submissions_failed WHERE prefix = $3 + SELECT id AS "id: SubmissionId" FROM submissions_failed WHERE prefix = $1 "#, - prefix, - prefix, prefix ) .fetch_optional(conn.get_inner()) diff --git a/opsqueue/src/consumer/dispatcher/metastate.rs b/opsqueue/src/consumer/dispatcher/metastate.rs index f96534c..9eb9993 100644 --- a/opsqueue/src/consumer/dispatcher/metastate.rs +++ b/opsqueue/src/consumer/dispatcher/metastate.rs @@ -1,4 +1,3 @@ -use crossbeam_skiplist::SkipSet; use dashmap::{DashMap, Entry}; use rustc_hash::FxBuildHasher; use tracing; @@ -70,7 +69,6 @@ pub type MetaStateVal = i64; #[derive(Debug, Default)] pub struct MetaStateField { vals_to_counts: DashMap, - counts_to_vals: SkipSet<(usize, MetaStateVal)>, } impl MetaStateField { @@ -80,55 +78,47 @@ impl MetaStateField { } fn increment(&self, val: MetaStateVal) { - match self.vals_to_counts.entry(val) { - Entry::Vacant(entry) => { - self.counts_to_vals.insert((1, *entry.key())); - entry.insert(1); - } - Entry::Occupied(mut entry) => { - // The entry is now locked, so we can also safely update the relevant element of the SkipSet - let count = entry.get(); - let mut set_entry = (*count, *entry.key()); - self.counts_to_vals.remove(&set_entry); - set_entry.0 += 1; - self.counts_to_vals.insert(set_entry); - *entry.get_mut() += 1; - } - } + self.vals_to_counts + .entry(val) + .and_modify(|count| *count += 1) + .or_insert(1); } fn decrement(&self, val: MetaStateVal) { - match self.vals_to_counts.entry(val) { - Entry::Vacant(_entry) => { - unreachable!() - } - Entry::Occupied(mut entry) => { - // The entry is now locked, so we can also safely update the relevant element of the SkipSet - let count = entry.get(); - let mut set_entry = (*count, *entry.key()); - if *count == 1 { - *entry.get_mut() -= 1; - self.counts_to_vals.remove(&set_entry); - entry.remove(); - } else { - *entry.get_mut() -= 1; - self.counts_to_vals.remove(&set_entry); - set_entry.0 -= 1; - self.counts_to_vals.insert(set_entry); - } - } + if let Entry::Occupied(entry) = self + .vals_to_counts + .entry(val) + .and_modify(|count| *count -= 1) + && *entry.get() == 0 + { + entry.remove(); } } + #[must_use] pub fn is_empty(&self) -> bool { self.vals_to_counts.is_empty() } - pub fn too_high_counts(&self, max: usize) -> impl Iterator + '_ { - tracing::debug!("metastate: {self:?}"); - self.counts_to_vals - .range((max, 0)..) - .map(|entry| entry.value().1) + #[must_use] + pub fn get(&self, val: &MetaStateVal) -> Option { + self.vals_to_counts.get(val).map(|count| *count) + } + + /// The whole value -> count map as a JSON object, for handing to `SQLite` + /// in a single FFI call. + #[must_use] + pub fn to_json(&self) -> String { + use std::fmt::Write as _; + let mut out = String::from("{"); + for entry in &self.vals_to_counts { + if out.len() > 1 { + out.push(','); + } + let _ = write!(out, "\"{}\":{}", entry.key(), entry.value()); + } + out.push('}'); + out } } @@ -158,14 +148,17 @@ mod tests { sut.increment(key, val); } - dbg!(&sut); - - let too_highs: Vec<_> = sut - .get(key) - .expect("Should exist at this stage") - .too_high_counts(group_size) - .collect(); - assert_eq!(too_highs.len(), n_groups); + { + // We have to release the selected state_field before we can decrement it, otherwise we + // would deadlock on the DashMap lock. + let state_field = sut.get(key).expect("Should exist at this stage"); + for group in 0..n_groups { + assert_eq!( + state_field.get(&i64::try_from(group).unwrap()), + Some(group_size) + ); + } + } // Decrement in a different order vals.shuffle(&mut rand::rng()); diff --git a/opsqueue/src/consumer/dispatcher/mod.rs b/opsqueue/src/consumer/dispatcher/mod.rs index 8ba9b03..728566b 100644 --- a/opsqueue/src/consumer/dispatcher/mod.rs +++ b/opsqueue/src/consumer/dispatcher/mod.rs @@ -3,15 +3,17 @@ pub mod reserver; use crate::{ common::{ - chunk::{Chunk, ChunkId}, - submission::Submission, + chunk::{Chunk, ChunkId, ChunkIndex}, + submission::{Submission, SubmissionId}, }, db::{Connection, Pool, ReaderPool, magic::Bool}, }; use futures::stream::{StreamExt as _, TryStreamExt as _}; +use libsqlite3_sys as ffi; use metastate::MetaState; use reserver::Reserver; use sqlx::QueryBuilder; +use std::ffi::CStr; use std::time::{Duration, Instant}; use tokio::sync::mpsc::UnboundedSender; use tokio_util::sync::CancellationToken; @@ -21,6 +23,167 @@ use std::sync::Arc; use super::strategy; use crate::common::StrategicMetadataMap; +unsafe extern "C" fn sqlite_reserved_chunk_lookup( + context: *mut ffi::sqlite3_context, + n_args: i32, + args: *mut *mut ffi::sqlite3_value, +) { + if n_args != 2 { + tracing::error!( + n_args, + "opsqueue_is_reserved called with unexpected argument count" + ); + // Fail open: this callback is an optimization only. + unsafe { ffi::sqlite3_result_int(context, 0) }; + return; + } + + let user_data = unsafe { ffi::sqlite3_user_data(context) } + .cast_const() + .cast::>(); + if user_data.is_null() { + tracing::error!("opsqueue_is_reserved called without registered reserver user_data"); + // Fail open: this callback is an optimization only. + unsafe { ffi::sqlite3_result_int(context, 0) }; + return; + } + + let submission_id_raw = unsafe { ffi::sqlite3_value_int64(*args.add(0)) }; + let chunk_index_raw = unsafe { ffi::sqlite3_value_int64(*args.add(1)) }; + + let Ok(submission_id) = SubmissionId::try_from(submission_id_raw) else { + tracing::error!( + submission_id_raw, + "opsqueue_is_reserved got invalid submission_id" + ); + // Fail open: this callback is an optimization only. + unsafe { ffi::sqlite3_result_int(context, 0) }; + return; + }; + let Ok(chunk_index) = ChunkIndex::try_from(chunk_index_raw) else { + tracing::error!( + chunk_index_raw, + "opsqueue_is_reserved got invalid chunk_index" + ); + // Fail open: this callback is an optimization only. + unsafe { ffi::sqlite3_result_int(context, 0) }; + return; + }; + + let chunk_id = ChunkId::from((submission_id, chunk_index)); + let is_reserved = unsafe { &*user_data }.is_reserved(&chunk_id); + unsafe { ffi::sqlite3_result_int(context, i32::from(is_reserved)) }; +} + +unsafe extern "C" fn sqlite_reserved_chunk_lookup_destructor(ptr: *mut std::ffi::c_void) { + if ptr.is_null() { + return; + } + let _boxed: Box> = unsafe { Box::from_raw(ptr.cast()) }; +} + +unsafe extern "C" fn sqlite_metadata_count_lookup( + context: *mut ffi::sqlite3_context, + n_args: i32, + args: *mut *mut ffi::sqlite3_value, +) { + if n_args != 2 { + tracing::error!( + n_args, + "opsqueue_metadata_count called with unexpected argument count" + ); + unsafe { ffi::sqlite3_result_null(context) }; + return; + } + + let user_data = unsafe { ffi::sqlite3_user_data(context) } + .cast_const() + .cast::>(); + if user_data.is_null() { + tracing::error!("opsqueue_metadata_count called without registered metastate user_data"); + unsafe { ffi::sqlite3_result_null(context) }; + return; + } + + let metadata_key_ptr = unsafe { ffi::sqlite3_value_text(*args.add(0)) }; + if metadata_key_ptr.is_null() { + unsafe { ffi::sqlite3_result_null(context) }; + return; + } + let Ok(metadata_key) = unsafe { CStr::from_ptr(metadata_key_ptr.cast()) }.to_str() else { + tracing::error!("opsqueue_metadata_count got non-utf8 metadata_key"); + unsafe { ffi::sqlite3_result_null(context) }; + return; + }; + + let metadata_value = unsafe { ffi::sqlite3_value_int64(*args.add(1)) }; + + if let Some(meta_count) = unsafe { &*user_data } + .get(metadata_key) + .and_then(|meta_keys| meta_keys.get(&metadata_value)) + { + unsafe { + ffi::sqlite3_result_int64(context, i64::try_from(meta_count).unwrap_or(i64::MAX)); + }; + } else { + unsafe { ffi::sqlite3_result_null(context) }; + } +} + +/// Returns the whole value -> count map for one metadata key as a JSON object, +/// so the caller can obtain every count in a single FFI call. +unsafe extern "C" fn sqlite_metadata_counts_lookup( + context: *mut ffi::sqlite3_context, + n_args: i32, + args: *mut *mut ffi::sqlite3_value, +) { + if n_args != 1 { + tracing::error!( + n_args, + "opsqueue_metadata_counts called with unexpected argument count" + ); + unsafe { ffi::sqlite3_result_null(context) }; + return; + } + + let user_data = unsafe { ffi::sqlite3_user_data(context) } + .cast_const() + .cast::>(); + if user_data.is_null() { + tracing::error!("opsqueue_metadata_counts called without registered metastate user_data"); + unsafe { ffi::sqlite3_result_null(context) }; + return; + } + + let metadata_key_ptr = unsafe { ffi::sqlite3_value_text(*args.add(0)) }; + if metadata_key_ptr.is_null() { + unsafe { ffi::sqlite3_result_null(context) }; + return; + } + let Ok(metadata_key) = unsafe { CStr::from_ptr(metadata_key_ptr.cast()) }.to_str() else { + tracing::error!("opsqueue_metadata_counts got non-utf8 metadata_key"); + unsafe { ffi::sqlite3_result_null(context) }; + return; + }; + + let json = match unsafe { &*user_data }.get(metadata_key) { + Some(field) => field.to_json(), + None => "{}".to_string(), + }; + let len = i32::try_from(json.len()).unwrap_or(i32::MAX); + unsafe { + // SQLITE_TRANSIENT tells SQLite to copy the bytes before we drop them. + ffi::sqlite3_result_text(context, json.as_ptr().cast(), len, ffi::SQLITE_TRANSIENT()); + }; +} + +unsafe extern "C" fn sqlite_metadata_count_lookup_destructor(ptr: *mut std::ffi::c_void) { + if ptr.is_null() { + return; + } + let _boxed: Box> = unsafe { Box::from_raw(ptr.cast()) }; +} + #[derive(Debug, Clone)] pub struct Dispatcher { reserver: Reserver, @@ -74,9 +237,10 @@ impl Dispatcher { stale_chunks_notifier: &UnboundedSender, ) -> Result, sqlx::Error> { let mut conn = pool.reader_conn().await?; + self.register_lookups(conn.get_inner()).await?; let mut query_builder = QueryBuilder::new(""); let stream = strategy - .build_query(&mut query_builder, &self.metastate) + .build_query(&mut query_builder) .build_query_as() .fetch(conn.get_inner()); stream @@ -89,6 +253,94 @@ impl Dispatcher { .await } + async fn register_lookups(&self, conn: &mut sqlx::SqliteConnection) -> Result<(), sqlx::Error> { + let mut handle = conn.lock_handle().await?; + let sqlite = handle.as_raw_handle().as_ptr(); + let reserved_function_name = b"opsqueue_is_reserved\0"; + let metadata_count_function_name = b"opsqueue_metadata_count\0"; + + // Register the current reserver state on this connection. + // Re-registering replaces any previous callback on this handle. + let user_data = Box::new(self.reserver.clone()); + let user_data = Box::into_raw(user_data).cast::(); + + let rc = unsafe { + ffi::sqlite3_create_function_v2( + sqlite, + reserved_function_name.as_ptr().cast(), + 2, + ffi::SQLITE_UTF8, + user_data, + Some(sqlite_reserved_chunk_lookup), + None, + None, + Some(sqlite_reserved_chunk_lookup_destructor), + ) + }; + + if rc != ffi::SQLITE_OK { + // We don't need to explicitly call the destructor. + // c.f. https://sqlite.org/c3ref/create_function.html + return Err(sqlx::Error::Protocol(format!( + "sqlite3_create_function_v2 failed with rc={rc}" + ))); + } + + // Register metadata count lookup backed by current metastate. + // Re-registering replaces any previous callback on this handle. + let user_data = Box::new(self.metastate.clone()); + let user_data = Box::into_raw(user_data).cast::(); + + let rc = unsafe { + ffi::sqlite3_create_function_v2( + sqlite, + metadata_count_function_name.as_ptr().cast(), + 2, + ffi::SQLITE_UTF8, + user_data, + Some(sqlite_metadata_count_lookup), + None, + None, + Some(sqlite_metadata_count_lookup_destructor), + ) + }; + + if rc != ffi::SQLITE_OK { + // We don't need to explicitly call the destructor. + // c.f. https://sqlite.org/c3ref/create_function.html + return Err(sqlx::Error::Protocol(format!( + "sqlite3_create_function_v2 failed with rc={rc}" + ))); + } + + // Register the bulk metadata counts lookup backed by current metastate. + let counts_function_name = b"opsqueue_metadata_counts\0"; + let user_data = Box::new(self.metastate.clone()); + let user_data = Box::into_raw(user_data).cast::(); + + let rc = unsafe { + ffi::sqlite3_create_function_v2( + sqlite, + counts_function_name.as_ptr().cast(), + 1, + ffi::SQLITE_UTF8, + user_data, + Some(sqlite_metadata_counts_lookup), + None, + None, + Some(sqlite_metadata_count_lookup_destructor), + ) + }; + + if rc != ffi::SQLITE_OK { + return Err(sqlx::Error::Protocol(format!( + "sqlite3_create_function_v2 failed with rc={rc}" + ))); + } + + Ok(()) + } + fn reserve_chunk( &self, chunk: Chunk, @@ -152,3 +404,58 @@ impl Dispatcher { .run_pending_tasks_periodically(cancellation_token); } } + +#[cfg(test)] +#[cfg(feature = "server-logic")] +mod test { + use super::*; + use crate::common::chunk::ChunkId; + use crate::common::chunk::ChunkSize; + use crate::db::DBPools; + use tokio::sync::mpsc::unbounded_channel; + use ux::u63; + + #[sqlx::test(migrator = "crate::MIGRATOR")] + async fn fetch_and_reserve_chunks_excludes_already_reserved(db: sqlx::SqlitePool) { + let pools = DBPools::from_test_pool(&db); + let dispatcher = Dispatcher::new(Duration::from_mins(1)); + let (stale_chunks_notifier, mut _stale_chunks_receiver) = unbounded_channel::(); + + let mut writer_conn = pools.writer_conn().await.unwrap(); + let submission_id = crate::common::submission::db::insert_submission_from_chunks( + None, + vec![Some("a".into()), Some("b".into()), Some("c".into())], + None, + StrategicMetadataMap::default(), + ChunkSize::default(), + &mut writer_conn, + ) + .await + .unwrap(); + + let pre_reserved_chunk = ChunkId::from((submission_id, u63::new(0).into())); + dispatcher + .reserver() + .try_reserve( + pre_reserved_chunk, + pre_reserved_chunk, + &stale_chunks_notifier, + ) + .expect("precondition: pre-reserving chunk should succeed"); + + let reserved = dispatcher + .fetch_and_reserve_chunks( + pools.reader_pool(), + strategy::Strategy::Oldest, + 10, + &stale_chunks_notifier, + ) + .await + .unwrap(); + + assert_eq!(reserved.len(), 2); + assert!(reserved.iter().all(|(chunk, _submission)| { + ChunkId::from((chunk.submission_id, chunk.chunk_index)) != pre_reserved_chunk + })); + } +} diff --git a/opsqueue/src/consumer/dispatcher/reserver.rs b/opsqueue/src/consumer/dispatcher/reserver.rs index 0699857..91b93e6 100644 --- a/opsqueue/src/consumer/dispatcher/reserver.rs +++ b/opsqueue/src/consumer/dispatcher/reserver.rs @@ -83,6 +83,11 @@ where } } + /// Returns whether a key currently has an active reservation. + pub fn is_reserved(&self, key: &K) -> bool { + self.reservations.contains_key(key) + } + /// Removes a particular key-val from the reserver. /// Afterwards, it is possible to reserve it again. /// diff --git a/opsqueue/src/consumer/strategy.rs b/opsqueue/src/consumer/strategy.rs index a725d13..c37ccd0 100644 --- a/opsqueue/src/consumer/strategy.rs +++ b/opsqueue/src/consumer/strategy.rs @@ -7,9 +7,6 @@ use sqlx::{QueryBuilder, Sqlite}; #[cfg(feature = "server-logic")] use crate::common::chunk::Chunk; -#[cfg(feature = "server-logic")] -use super::dispatcher::metastate::MetaState; - #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub enum Strategy { Oldest, @@ -21,75 +18,190 @@ pub enum Strategy { }, } +/// Iterator over the `meta_key`s of a chain of nested +/// [`Strategy::PreferDistinct`]. outermost first. Stops at the first +/// non-`PreferDistinct` strategy, which can afterwards be retrieved with +/// [`MetaKeysIter::take`]. +pub struct MetaKeysIter<'a> { + strategy: &'a Strategy, +} + +impl<'a> MetaKeysIter<'a> { + /// The first non-[`Strategy::PreferDistinct`] strategy in the chain. + #[must_use] + pub fn take(self) -> &'a Strategy { + self.strategy + } +} + +impl<'a> Iterator for MetaKeysIter<'a> { + type Item = &'a str; + + fn next(&mut self) -> Option { + match self.strategy { + Strategy::Oldest | Strategy::Newest | Strategy::Random => None, + Strategy::PreferDistinct { + meta_key, + underlying, + } => { + self.strategy = underlying.as_ref(); + Some(meta_key.as_str()) + } + } + } +} + +impl Strategy { + /// Iterate over the `meta_key`s of this chain of nested + /// [`Strategy::PreferDistinct`], outermost first. + #[must_use] + pub fn meta_keys(&self) -> MetaKeysIter<'_> { + MetaKeysIter { strategy: self } + } +} + #[cfg(feature = "server-logic")] impl Strategy { pub fn build_query<'a>( &'a self, qb: &'a mut QueryBuilder, - metastate: &MetaState, ) -> &'a mut QueryBuilder { - let qb = self.build_query_snippet(qb, metastate); + let qb = self.build_query_snippet_returning_chunks(qb); tracing::trace!("sql: {:?}", qb.sql()); qb } - fn build_query_snippet<'a>( + fn build_query_snippet_returning_chunks<'a>( &'a self, qb: &'a mut QueryBuilder, - metastate: &MetaState, ) -> &'a mut QueryBuilder { use Strategy::{Newest, Oldest, PreferDistinct, Random}; + let ffi_is_reserved = "opsqueue_is_reserved(chunks.submission_id, chunks.chunk_index) = 0"; match self { - Oldest => qb.push("SELECT * FROM chunks ORDER BY submission_id ASC"), - Newest => qb.push("SELECT * FROM chunks ORDER BY submission_id DESC"), - Random => { - let random_offset: u16 = rand::random(); - qb.push("SELECT * FROM chunks WHERE random_order >= ") - .push_bind(random_offset) - .push(" UNION ALL SELECT * FROM chunks WHERE random_order < ") - .push_bind(random_offset) + Oldest => qb + .push("SELECT * FROM chunks") + .push(format!(" WHERE {ffi_is_reserved}")) + .push(" ORDER BY submission_id ASC"), + Newest => qb + .push("SELECT * FROM chunks") + .push(format!(" WHERE {ffi_is_reserved}")) + .push(" ORDER BY submission_id DESC"), + Random => Self::push_random_order_query(qb, "*", "chunks", Some(ffi_is_reserved)), + PreferDistinct { .. } => { + // Unique submission IDs from the underlying strategy. + let qb = qb.push("WITH underlying_submission_ids AS MATERIALIZED ("); + let qb = self.build_query_snippet_returning_submission_ids(qb); + qb.push(") "); + // In SQLite, CROSS JOIN ON/WHERE does NOT produce N + // x M rows, it acts as an INNER JOIN but forces the query + // planner to use '' as the outer loop, preserving the + // underlying sort order. + // c.f. https://sqlite.org/optoverview.html#manual_control_of_query_plans_using_cross_join + qb.push(format!( + " SELECT chunks.* + FROM underlying_submission_ids + CROSS JOIN chunks + ON chunks.submission_id = underlying_submission_ids.submission_id + AND {ffi_is_reserved}", + )) } + } + } - PreferDistinct { - meta_key, - underlying, - } => { - let qb = qb.push(format_args!("WITH inner_{meta_key} AS NOT MATERIALIZED (")); - let qb = underlying.build_query_snippet(qb, metastate); - qb.push(format_args!( - r"), - taken_{meta_key} AS ( - SELECT * FROM submissions_metadata - WHERE - submissions_metadata.metadata_key = ", - )); - qb.push_bind(meta_key); - qb.push( - r" AND submissions_metadata.metadata_value IN (SELECT value FROM json_each(", - ); - match metastate.get(meta_key) { - None => { - tracing::trace!("No metastatefield for key: {meta_key}"); - } - Some(field) => { - let taken_values: Vec<_> = field.too_high_counts(1).collect(); - let taken_values_string = - serde_json::to_string(&taken_values).expect("Always valid JSON"); - tracing::trace!( - "Taken values that are left out of PreferDistinct: {taken_values_string:?}" - ); - qb.push_bind(taken_values_string); - } + fn build_query_snippet_returning_submission_ids<'a>( + &'a self, + qb: &'a mut QueryBuilder, + ) -> &'a mut QueryBuilder { + use Strategy::{Newest, Oldest, PreferDistinct, Random}; + match self { + Oldest => qb.push("SELECT id as submission_id FROM submissions ORDER BY id ASC"), + Newest => qb.push("SELECT id as submission_id FROM submissions ORDER BY id DESC"), + Random => Self::push_random_order_query(qb, "id as submission_id", "submissions", None), + PreferDistinct { .. } => { + let mut meta_keys_iter = self.meta_keys(); + let meta_keys: Vec<&str> = meta_keys_iter.by_ref().collect(); + let underlying = meta_keys_iter.take(); + + // Unique submission IDs from the underlying strategy. Note how + // we also keep the row number from the underlying query, this + // is used as a tie-breaker if metadata counts are equal. + let qb = qb.push("WITH inner AS NOT MATERIALIZED ("); + qb.push("SELECT submission_id, ROW_NUMBER() OVER () as underlying_row FROM ( "); + let qb = underlying.build_query_snippet_returning_submission_ids(qb); + qb.push(" ))"); + + // In-flight chunk count per submission, per meta key. + // + // The FFI call returns all counts as JSON in a single call. + // The CROSS JOIN ON ensures the json_each is the outer loop, + // and only performed once. + for (i, meta_key) in meta_keys.iter().enumerate() { + qb.push(format!( + ", counts_{i} AS ( + SELECT sm.submission_id, ffi_counts.value AS count + FROM json_each(opsqueue_metadata_counts(" + )); + qb.push_bind(*meta_key); + qb.push( + ")) + ffi_counts + CROSS JOIN submissions_metadata sm + ON sm.metadata_value = CAST(ffi_counts.key AS INTEGER) + WHERE sm.metadata_key = ", + ); + qb.push_bind(*meta_key); + qb.push(")"); } - qb.push(format_args!(")) - ) - SELECT * FROM inner_{meta_key} WHERE NOT EXISTS (SELECT 1 FROM taken_{meta_key} WHERE inner_{meta_key}.submission_id = taken_{meta_key}.submission_id) - UNION ALL - SELECT * FROM inner_{meta_key} WHERE EXISTS (SELECT 1 FROM taken_{meta_key} WHERE inner_{meta_key}.submission_id = taken_{meta_key}.submission_id) - ")) + + // Submissions ranked by in-flight chunks. Submissions without a + // value for a key get a NULL count and so are ranked first. + qb.push(" SELECT inner.submission_id FROM inner"); + for i in 0..meta_keys.len() { + qb.push(format!( + " LEFT JOIN counts_{i} ON inner.submission_id = counts_{i}.submission_id" + )); + } + for i in 0..meta_keys.len() { + qb.push(if i == 0 { " ORDER BY " } else { ", " }); + qb.push(format!("counts_{i}.count ASC NULLS FIRST")); + } + + if meta_keys.is_empty() { + panic!("`PreferDistinct` always yields at least one meta key.") + } else { + // Ensure that submissions with equal metadata counts use the + // ordering of the underlying strategy as a tie-breaker. + qb.push(", inner.underlying_row ASC"); + } + + qb } } } + + /// Append a query snippet to select from the `random_order` column on the + /// given table using the "cutting the deck" technique. + fn push_random_order_query<'a>( + qb: &'a mut QueryBuilder, + columns: &str, + table_name: &str, + condition: Option<&str>, + ) -> &'a mut QueryBuilder { + let random_offset: u16 = rand::random(); + let push_select = |qb: &mut QueryBuilder, operator: &str| { + qb.push(format!( + "SELECT {columns} FROM {table_name} WHERE random_order {operator} " + )) + .push_bind(random_offset); + if let Some(condition_) = condition { + qb.push(format!(" AND {condition_}")); + } + }; + push_select(qb, ">="); + qb.push(" UNION ALL "); + push_select(qb, "<"); + qb + } } #[cfg(feature = "server-logic")] @@ -98,22 +210,107 @@ pub type ChunkStream<'a> = BoxStream<'a, Result>; #[cfg(test)] #[cfg(feature = "server-logic")] pub mod test { + use super::*; use crate::common::StrategicMetadataMap; use crate::common::chunk::ChunkSize; - - use super::*; use itertools::Itertools; + use libsqlite3_sys as ffi; use sqlformat::{FormatOptions, QueryParams, format}; use sqlx::Row; use sqlx::{QueryBuilder, Sqlite, SqliteConnection}; + unsafe extern "C" fn sqlite_reserved_chunk_lookup_noop( + context: *mut ffi::sqlite3_context, + _n_args: i32, + _args: *mut *mut ffi::sqlite3_value, + ) { + unsafe { ffi::sqlite3_result_int(context, 0) }; + } + + unsafe extern "C" fn sqlite_metadata_counts_lookup_noop( + context: *mut ffi::sqlite3_context, + _n_args: i32, + _args: *mut *mut ffi::sqlite3_value, + ) { + unsafe { + ffi::sqlite3_result_text(context, c"{}".as_ptr(), 2, ffi::SQLITE_TRANSIENT()); + }; + } + + unsafe extern "C" fn sqlite_metadata_count_lookup_noop( + context: *mut ffi::sqlite3_context, + _n_args: i32, + _args: *mut *mut ffi::sqlite3_value, + ) { + unsafe { ffi::sqlite3_result_null(context) }; + } + + async fn register_lookup_noops(conn: &mut SqliteConnection) { + let mut handle = conn.lock_handle().await.unwrap(); + let sqlite = handle.as_raw_handle().as_ptr(); + let function_name = b"opsqueue_is_reserved\0"; + let rc = unsafe { + ffi::sqlite3_create_function_v2( + sqlite, + function_name.as_ptr().cast(), + 2, + ffi::SQLITE_UTF8, + std::ptr::null_mut(), + Some(sqlite_reserved_chunk_lookup_noop), + None, + None, + None, + ) + }; + assert_eq!(rc, ffi::SQLITE_OK, "register opsqueue_is_reserved failed"); + + let function_name = b"opsqueue_metadata_count\0"; + let rc = unsafe { + ffi::sqlite3_create_function_v2( + sqlite, + function_name.as_ptr().cast(), + 2, + ffi::SQLITE_UTF8, + std::ptr::null_mut(), + Some(sqlite_metadata_count_lookup_noop), + None, + None, + None, + ) + }; + assert_eq!( + rc, + ffi::SQLITE_OK, + "register opsqueue_metadata_count failed" + ); + + let function_name = b"opsqueue_metadata_counts\0"; + let rc = unsafe { + ffi::sqlite3_create_function_v2( + sqlite, + function_name.as_ptr().cast(), + 1, + ffi::SQLITE_UTF8, + std::ptr::null_mut(), + Some(sqlite_metadata_counts_lookup_noop), + None, + None, + None, + ) + }; + assert_eq!( + rc, + ffi::SQLITE_OK, + "register opsqueue_metadata_counts failed" + ); + } + async fn explain(qb: &mut sqlx::QueryBuilder, conn: &mut SqliteConnection) -> String { let formatted_query = format( qb.sql().as_str(), &QueryParams::None, &FormatOptions::default(), ); - sqlx::raw_sql(sqlx::AssertSqlSafe(format!( "EXPLAIN QUERY PLAN {formatted_query}" ))) @@ -143,13 +340,34 @@ pub mod test { ); } + /// A weaker version of `assert_streaming_query`, for `PreferDistinct`. + /// + /// `PreferDistinct` cannot stream: to rank submissions by how many of their + /// chunks are already in flight, it has to sort the `submissions` table. We + /// accept that cost, because there are fewer submissions than chunks. + /// + /// What we do not accept is doing the same to `chunks`, so we only require + /// that `chunks` is reached by an index seek. + fn assert_streaming_chunks(qb: &sqlx::QueryBuilder, explained: &str) { + let query_binding = qb.sql(); + let query = query_binding.as_str(); + assert!( + !explained.contains("SCAN chunks"), + "Query should never scan the whole `chunks` backlog, but it did.\n\nQuery: {query}\n\nPlan: \n\n{explained}" + ); + assert!( + explained.contains("SEARCH chunks"), + "Query should reach `chunks` via an index seek, but it did not.\n\nQuery: {query}\n\nPlan: \n\n{explained}" + ); + } + #[sqlx::test(migrator = "crate::MIGRATOR")] pub async fn test_query_plan_oldest(db: sqlx::SqlitePool) { let mut conn = db.acquire().await.unwrap(); + register_lookup_noops(&mut conn).await; let mut qb = QueryBuilder::new(""); - let metastate = MetaState::default(); - let qb = Strategy::Oldest.build_query(&mut qb, &metastate); + let qb = Strategy::Oldest.build_query(&mut qb); let options = FormatOptions::default(); let formatted_query = format(qb.sql().as_str(), &QueryParams::None, &options); insta::assert_snapshot!(formatted_query, @" @@ -157,6 +375,8 @@ pub mod test { * FROM chunks + WHERE + opsqueue_is_reserved(chunks.submission_id, chunks.chunk_index) = 0 ORDER BY submission_id ASC "); @@ -169,10 +389,10 @@ pub mod test { #[sqlx::test(migrator = "crate::MIGRATOR")] pub async fn test_query_plan_newest(db: sqlx::SqlitePool) { let mut conn = db.acquire().await.unwrap(); + register_lookup_noops(&mut conn).await; let mut qb = QueryBuilder::new(""); - let metastate = MetaState::default(); - let qb = Strategy::Newest.build_query(&mut qb, &metastate); + let qb = Strategy::Newest.build_query(&mut qb); let options = FormatOptions::default(); let formatted_query = format(qb.sql().as_str(), &QueryParams::None, &options); insta::assert_snapshot!(formatted_query, @" @@ -180,6 +400,8 @@ pub mod test { * FROM chunks + WHERE + opsqueue_is_reserved(chunks.submission_id, chunks.chunk_index) = 0 ORDER BY submission_id DESC "); @@ -192,10 +414,10 @@ pub mod test { #[sqlx::test(migrator = "crate::MIGRATOR")] pub async fn test_query_plan_random(db: sqlx::SqlitePool) { let mut conn = db.acquire().await.unwrap(); - let metastate = MetaState::default(); + register_lookup_noops(&mut conn).await; let mut qb = QueryBuilder::new(""); - let qb = Strategy::Random.build_query(&mut qb, &metastate); + let qb = Strategy::Random.build_query(&mut qb); let formatted_query = format( qb.sql().as_str(), @@ -209,6 +431,7 @@ pub mod test { chunks WHERE random_order >= ? + AND opsqueue_is_reserved(chunks.submission_id, chunks.chunk_index) = 0 UNION ALL SELECT * @@ -216,6 +439,7 @@ pub mod test { chunks WHERE random_order < ? + AND opsqueue_is_reserved(chunks.submission_id, chunks.chunk_index) = 0 "); let explained = explain(qb, &mut conn).await; @@ -224,8 +448,8 @@ pub mod test { 1, 0, COMPOUND QUERY 2, 1, LEFT-MOST SUBQUERY 5, 2, SEARCH chunks USING INDEX random_chunks_order (random_order>?) - 22, 1, UNION ALL - 25, 22, SEARCH chunks USING INDEX random_chunks_order (random_order= ? - UNION ALL - SELECT - * - FROM - chunks - WHERE - random_order < ? - ), - taken_company_id AS ( - SELECT - * - FROM - submissions_metadata - WHERE - submissions_metadata.metadata_key = ? - AND submissions_metadata.metadata_value IN ( - SELECT - value - FROM - json_each() - ) - ) - SELECT - * - FROM - inner_company_id - WHERE - NOT EXISTS ( + underlying_submission_ids AS MATERIALIZED ( + WITH + inner AS NOT MATERIALIZED ( SELECT - 1 + submission_id, + ROW_NUMBER() OVER () as underlying_row FROM - taken_company_id - WHERE - inner_company_id.submission_id = taken_company_id.submission_id - ) - UNION ALL - SELECT - * - FROM - inner_company_id - WHERE - EXISTS ( + ( + SELECT + id as submission_id + FROM + submissions + WHERE + random_order >= ? + UNION ALL + SELECT + id as submission_id + FROM + submissions + WHERE + random_order < ? + ) + ), + counts_0 AS ( SELECT - 1 + sm.submission_id, + ffi_counts.value AS count FROM - taken_company_id + json_each(opsqueue_metadata_counts(?)) ffi_counts + CROSS JOIN submissions_metadata sm ON sm.metadata_value = CAST(ffi_counts.key AS INTEGER) WHERE - inner_company_id.submission_id = taken_company_id.submission_id + sm.metadata_key = ? ) + SELECT + inner.submission_id + FROM + inner + LEFT JOIN counts_0 ON inner.submission_id = counts_0.submission_id + ORDER BY + counts_0.count ASC NULLS FIRST, + inner.underlying_row ASC + ) + SELECT + chunks.* + FROM + underlying_submission_ids + CROSS JOIN chunks ON chunks.submission_id = underlying_submission_ids.submission_id + AND opsqueue_is_reserved(chunks.submission_id, chunks.chunk_index) = 0 "); let explained = explain(qb, &mut conn).await; - assert_streaming_query(qb, &explained); + assert_streaming_chunks(qb, &explained); insta::assert_snapshot!(explained, @" - 1, 0, COMPOUND QUERY - 2, 1, LEFT-MOST SUBQUERY - 3, 2, COMPOUND QUERY - 4, 3, LEFT-MOST SUBQUERY - 7, 4, SEARCH chunks USING INDEX random_chunks_order (random_order>?) - 16, 4, CORRELATED SCALAR SUBQUERY 5 - 20, 16, SEARCH submissions_metadata USING PRIMARY KEY (submission_id=? AND metadata_key=?) - 30, 16, LIST SUBQUERY 3 - 33, 30, SCAN json_each VIRTUAL TABLE INDEX 0: - 41, 30, CREATE BLOOM FILTER - 62, 3, UNION ALL - 65, 62, SEARCH chunks USING INDEX random_chunks_order (random_order?) - 121, 109, CORRELATED SCALAR SUBQUERY 7 - 125, 121, SEARCH submissions_metadata USING PRIMARY KEY (submission_id=? AND metadata_key=?) - 133, 121, REUSE LIST SUBQUERY 3 - 153, 108, UNION ALL - 156, 153, SEARCH chunks USING INDEX random_chunks_order (random_order?) + 22, 9, UNION ALL + 25, 22, SEARCH submissions USING INDEX random_submissions_order (random_order= ? - UNION ALL + ( + SELECT + id as submission_id + FROM + submissions + WHERE + random_order >= ? + UNION ALL + SELECT + id as submission_id + FROM + submissions + WHERE + random_order < ? + ) + ), + counts_0 AS ( SELECT - * + sm.submission_id, + ffi_counts.value AS count FROM - chunks + json_each(opsqueue_metadata_counts(?)) ffi_counts + CROSS JOIN submissions_metadata sm ON sm.metadata_value = CAST(ffi_counts.key AS INTEGER) WHERE - random_order < ? + sm.metadata_key = ? ), - taken_priority AS ( + counts_1 AS ( SELECT - * + sm.submission_id, + ffi_counts.value AS count FROM - submissions_metadata + json_each(opsqueue_metadata_counts(?)) ffi_counts + CROSS JOIN submissions_metadata sm ON sm.metadata_value = CAST(ffi_counts.key AS INTEGER) WHERE - submissions_metadata.metadata_key = ? - AND submissions_metadata.metadata_value IN ( - SELECT - value - FROM - json_each() - ) + sm.metadata_key = ? ) SELECT - * + inner.submission_id FROM - inner_priority - WHERE - NOT EXISTS ( - SELECT - 1 - FROM - taken_priority - WHERE - inner_priority.submission_id = taken_priority.submission_id - ) - UNION ALL - SELECT - * - FROM - inner_priority - WHERE - EXISTS ( - SELECT - 1 - FROM - taken_priority - WHERE - inner_priority.submission_id = taken_priority.submission_id - ) - ), - taken_company_id AS ( - SELECT - * - FROM - submissions_metadata - WHERE - submissions_metadata.metadata_key = ? - AND submissions_metadata.metadata_value IN ( - SELECT - value - FROM - json_each() - ) + inner + LEFT JOIN counts_0 ON inner.submission_id = counts_0.submission_id + LEFT JOIN counts_1 ON inner.submission_id = counts_1.submission_id + ORDER BY + counts_0.count ASC NULLS FIRST, + counts_1.count ASC NULLS FIRST, + inner.underlying_row ASC ) SELECT - * - FROM - inner_company_id - WHERE - NOT EXISTS ( - SELECT - 1 - FROM - taken_company_id - WHERE - inner_company_id.submission_id = taken_company_id.submission_id - ) - UNION ALL - SELECT - * + chunks.* FROM - inner_company_id - WHERE - EXISTS ( - SELECT - 1 - FROM - taken_company_id - WHERE - inner_company_id.submission_id = taken_company_id.submission_id - ) + underlying_submission_ids + CROSS JOIN chunks ON chunks.submission_id = underlying_submission_ids.submission_id + AND opsqueue_is_reserved(chunks.submission_id, chunks.chunk_index) = 0 "); let explained = explain(qb, &mut conn).await; - assert_streaming_query(qb, &explained); + assert_streaming_chunks(qb, &explained); insta::assert_snapshot!(explained, @" - 1, 0, COMPOUND QUERY - 2, 1, LEFT-MOST SUBQUERY - 3, 2, COMPOUND QUERY - 4, 3, LEFT-MOST SUBQUERY - 5, 4, COMPOUND QUERY - 6, 5, LEFT-MOST SUBQUERY - 9, 6, SEARCH chunks USING INDEX random_chunks_order (random_order>?) - 18, 6, CORRELATED SCALAR SUBQUERY 5 - 22, 18, SEARCH submissions_metadata USING PRIMARY KEY (submission_id=? AND metadata_key=?) - 32, 18, LIST SUBQUERY 3 - 35, 32, SCAN json_each VIRTUAL TABLE INDEX 0: - 43, 32, CREATE BLOOM FILTER - 56, 6, CORRELATED SCALAR SUBQUERY 11 - 60, 56, SEARCH submissions_metadata USING PRIMARY KEY (submission_id=? AND metadata_key=?) - 70, 56, LIST SUBQUERY 9 - 73, 70, SCAN json_each VIRTUAL TABLE INDEX 0: - 81, 70, CREATE BLOOM FILTER - 102, 5, UNION ALL - 105, 102, SEARCH chunks USING INDEX random_chunks_order (random_order?) - 213, 201, CORRELATED SCALAR SUBQUERY 7 - 217, 213, SEARCH submissions_metadata USING PRIMARY KEY (submission_id=? AND metadata_key=?) - 227, 213, LIST SUBQUERY 3 - 230, 227, SCAN json_each VIRTUAL TABLE INDEX 0: - 238, 227, CREATE BLOOM FILTER - 251, 201, CORRELATED SCALAR SUBQUERY 11 - 255, 251, SEARCH submissions_metadata USING PRIMARY KEY (submission_id=? AND metadata_key=?) - 265, 251, LIST SUBQUERY 9 - 268, 265, SCAN json_each VIRTUAL TABLE INDEX 0: - 276, 265, CREATE BLOOM FILTER - 297, 200, UNION ALL - 300, 297, SEARCH chunks USING INDEX random_chunks_order (random_order?) - 410, 398, CORRELATED SCALAR SUBQUERY 5 - 414, 410, SEARCH submissions_metadata USING PRIMARY KEY (submission_id=? AND metadata_key=?) - 424, 410, LIST SUBQUERY 3 - 427, 424, SCAN json_each VIRTUAL TABLE INDEX 0: - 435, 424, CREATE BLOOM FILTER - 448, 398, CORRELATED SCALAR SUBQUERY 13 - 452, 448, SEARCH submissions_metadata USING PRIMARY KEY (submission_id=? AND metadata_key=?) - 462, 448, LIST SUBQUERY 9 - 465, 462, SCAN json_each VIRTUAL TABLE INDEX 0: - 473, 462, CREATE BLOOM FILTER - 494, 397, UNION ALL - 497, 494, SEARCH chunks USING INDEX random_chunks_order (random_order?) - 605, 593, CORRELATED SCALAR SUBQUERY 7 - 609, 605, SEARCH submissions_metadata USING PRIMARY KEY (submission_id=? AND metadata_key=?) - 619, 605, LIST SUBQUERY 3 - 622, 619, SCAN json_each VIRTUAL TABLE INDEX 0: - 630, 619, CREATE BLOOM FILTER - 643, 593, CORRELATED SCALAR SUBQUERY 13 - 647, 643, SEARCH submissions_metadata USING PRIMARY KEY (submission_id=? AND metadata_key=?) - 657, 643, LIST SUBQUERY 9 - 660, 657, SCAN json_each VIRTUAL TABLE INDEX 0: - 668, 657, CREATE BLOOM FILTER - 689, 592, UNION ALL - 692, 689, SEARCH chunks USING INDEX random_chunks_order (random_order?) + 22, 9, UNION ALL + 25, 22, SEARCH submissions USING INDEX random_submissions_order (random_order = Strategy::Random - .build_query(&mut query_builder, &MetaState::default()) + .build_query(&mut query_builder) .build_query_as() .fetch(conn.get_inner()) .try_collect() @@ -798,7 +866,7 @@ pub mod test { let mut query_builder = QueryBuilder::default(); let vals2: Vec = Strategy::Random - .build_query(&mut query_builder, &MetaState::default()) + .build_query(&mut query_builder) .build_query_as() .fetch(conn.get_inner()) .try_collect() diff --git a/workspace-hack/Cargo.toml b/workspace-hack/Cargo.toml index c4bae4e..3dc7427 100644 --- a/workspace-hack/Cargo.toml +++ b/workspace-hack/Cargo.toml @@ -15,10 +15,8 @@ publish = false ### BEGIN HAKARI SECTION [dependencies] -aws-lc-rs = { version = "1", default-features = false, features = ["aws-lc-sys", "prebuilt-nasm"] } base64 = { version = "0.22" } chrono = { version = "0.4", features = ["serde"] } -crossbeam-epoch = { version = "0.9" } crossbeam-utils = { version = "0.8" } either = { version = "1", features = ["serde", "use_std"] } event-listener = { version = "5" } @@ -28,7 +26,7 @@ futures-io = { version = "0.3" } futures-sink = { version = "0.3" } futures-util = { version = "0.3", features = ["channel", "io", "sink"] } hyper = { version = "1", features = ["client", "http1", "http2", "server"] } -libsqlite3-sys = { version = "0.30", default-features = false, features = ["bundled", "pkg-config", "unlock_notify", "vcpkg"] } +libsqlite3-sys = { version = "0.30", features = ["bundled", "unlock_notify"] } log = { version = "0.4", default-features = false, features = ["std"] } num-traits = { version = "0.2", default-features = false, features = ["std"] } opentelemetry = { version = "0.32" } @@ -45,7 +43,7 @@ serde_json = { version = "1", features = ["raw_value"] } sha2 = { version = "0.10" } slab = { version = "0.4" } smallvec = { version = "1", default-features = false, features = ["const_new"] } -sqlx-core = { version = "0.9", features = ["_rt-tokio", "_tls-rustls-aws-lc-rs", "any", "chrono", "json", "migrate", "offline"] } +sqlx-core = { version = "0.9", features = ["_rt-tokio", "any", "chrono", "json", "migrate", "offline"] } sqlx-sqlite = { version = "0.9", default-features = false, features = ["any", "bundled", "chrono", "deserialize", "json", "load-extension", "migrate", "offline", "unlock-notify"] } thiserror = { version = "2" } tokio = { version = "1", features = ["fs", "io-util", "macros", "net", "rt-multi-thread", "signal", "sync", "time"] } @@ -58,7 +56,6 @@ uuid = { version = "1", features = ["fast-rng", "serde", "v4", "v7"] } zerocopy = { version = "0.8", default-features = false, features = ["derive", "simd"] } [build-dependencies] -aws-lc-rs = { version = "1", default-features = false, features = ["aws-lc-sys", "prebuilt-nasm"] } base64 = { version = "0.22" } chrono = { version = "0.4", features = ["serde"] } crossbeam-utils = { version = "0.8" } @@ -68,17 +65,16 @@ futures-channel = { version = "0.3", features = ["sink"] } futures-io = { version = "0.3" } futures-sink = { version = "0.3" } futures-util = { version = "0.3", features = ["channel", "io", "sink"] } -libsqlite3-sys = { version = "0.30", default-features = false, features = ["bundled", "pkg-config", "unlock_notify", "vcpkg"] } +libsqlite3-sys = { version = "0.30", features = ["bundled", "unlock_notify"] } log = { version = "0.4", default-features = false, features = ["std"] } num-traits = { version = "0.2", default-features = false, features = ["std"] } -rustls-pki-types = { version = "1", features = ["std"] } serde = { version = "1", features = ["alloc", "derive", "rc"] } serde_core = { version = "1", features = ["alloc", "rc"] } serde_json = { version = "1", features = ["raw_value"] } sha2 = { version = "0.10" } slab = { version = "0.4" } smallvec = { version = "1", default-features = false, features = ["const_new"] } -sqlx-core = { version = "0.9", features = ["_rt-tokio", "_tls-rustls-aws-lc-rs", "any", "chrono", "json", "migrate", "offline"] } +sqlx-core = { version = "0.9", features = ["_rt-tokio", "any", "chrono", "json", "migrate", "offline"] } sqlx-sqlite = { version = "0.9", default-features = false, features = ["any", "bundled", "chrono", "deserialize", "json", "load-extension", "migrate", "offline", "unlock-notify"] } syn = { version = "3", features = ["full", "visit-mut"] } thiserror = { version = "2" }