From 4c730fa12ad44ae4f11959aae3e3af00e5636cc6 Mon Sep 17 00:00:00 2001 From: Mike Hilgendorf Date: Tue, 30 Jun 2026 16:22:16 -0500 Subject: [PATCH] feat: object outbox Adds a durable outbox for object indexing and tests. Implemented for memory, lmdb, and scylla stores. --- packages/cli/test.nu | 9 +- ...ll_nested_survives_interrupted_indexing.nu | 54 ++++ .../pull_survives_interrupted_indexing.nu | 53 ++++ packages/cli/tests/index/outbox.nu | 34 +++ packages/index/src/batch.rs | 20 +- packages/index/src/cache/put.rs | 2 +- packages/index/src/grant/delete.rs | 2 +- packages/index/src/grant/put.rs | 2 +- packages/index/src/group/member/delete.rs | 2 +- packages/index/src/group/member/put.rs | 2 +- packages/index/src/group/put.rs | 2 +- packages/index/src/object/put.rs | 2 +- .../index/src/organization/member/delete.rs | 2 +- packages/index/src/organization/member/put.rs | 2 +- packages/index/src/organization/put.rs | 2 +- packages/index/src/process/put.rs | 2 +- packages/index/src/sandbox/put.rs | 2 +- packages/index/src/tag/put.rs | 2 +- packages/index/src/user/put.rs | 2 +- packages/server/src/checkin/artifact.rs | 20 +- packages/server/src/checkin/index.rs | 7 +- packages/server/src/config.rs | 28 ++ packages/server/src/index.rs | 246 +++++++++++++++++- packages/server/src/object/batch.rs | 82 +++--- packages/server/src/object/put.rs | 96 +++---- packages/server/src/object/store.rs | 141 +++++++++- packages/server/src/process/finalize.rs | 23 +- packages/server/src/process/finish.rs | 4 +- packages/server/src/process/put.rs | 22 +- packages/server/src/process/spawn.rs | 26 +- packages/server/src/sync/get.rs | 3 + packages/server/src/sync/get/index.rs | 63 +++++ packages/server/src/write.rs | 24 +- packages/stores/object/src/lmdb.rs | 20 +- packages/stores/object/src/lmdb/outbox.rs | 211 +++++++++++++++ .../stores/object/src/lmdb/outbox/object.rs | 183 +++++++++++++ packages/stores/object/src/memory.rs | 5 + packages/stores/object/src/memory/outbox.rs | 68 +++++ .../stores/object/src/memory/outbox/object.rs | 90 +++++++ packages/stores/object/src/scylla.cql | 7 + packages/stores/object/src/scylla.rs | 77 ++++++ packages/stores/object/src/scylla/outbox.rs | 51 ++++ .../stores/object/src/scylla/outbox/object.rs | 114 ++++++++ scripts/cloud/kubernetes.yaml | 2 + 44 files changed, 1581 insertions(+), 230 deletions(-) create mode 100644 packages/cli/tests/grants/pull_nested_survives_interrupted_indexing.nu create mode 100644 packages/cli/tests/grants/pull_survives_interrupted_indexing.nu create mode 100644 packages/cli/tests/index/outbox.nu create mode 100644 packages/stores/object/src/lmdb/outbox.rs create mode 100644 packages/stores/object/src/lmdb/outbox/object.rs create mode 100644 packages/stores/object/src/memory/outbox.rs create mode 100644 packages/stores/object/src/memory/outbox/object.rs create mode 100644 packages/stores/object/src/scylla/outbox.rs create mode 100644 packages/stores/object/src/scylla/outbox/object.rs diff --git a/packages/cli/test.nu b/packages/cli/test.nu index 51c2c68a0a..e7f91dae71 100644 --- a/packages/cli/test.nu +++ b/packages/cli/test.nu @@ -1025,7 +1025,14 @@ export def --env spawn [ } # Write the config. - let config = $default_config | merge deep --strategy append ($config | default {}) + let user_config = $config | default {} + let config = $default_config | merge deep --strategy append $user_config + # Replace the object store wholesale when the user provides one, so default lmdb fields are not deep-merged into another store kind. + let config = if ($user_config.object?.store? | is-not-empty) { + $config | upsert object.store $user_config.object.store + } else { + $config + } let config_path = mktemp -d let config_path = $config_path | path join 'config.json' $config | to json | save -f $config_path diff --git a/packages/cli/tests/grants/pull_nested_survives_interrupted_indexing.nu b/packages/cli/tests/grants/pull_nested_survives_interrupted_indexing.nu new file mode 100644 index 0000000000..0ca27143dc --- /dev/null +++ b/packages/cli/tests/grants/pull_nested_survives_interrupted_indexing.nu @@ -0,0 +1,54 @@ +use ../../test.nu * + +# A grant for a nested object survives the server being killed and authorizes a later pull of that nested object once a second server with the indexer drains the outbox. + +# The remote directory is shared by the producer and indexer phases. +let directory = mktemp -d + +# Start the producer with the indexer disabled so the push only enqueues the root grant to the outbox. +let producer = spawn --name producer --directory $directory --config { + authentication: { providers: { insecure: true } }, + indexer: { outbox: { partition_start: 0, partition_end: 0 } }, +} + +let alice = tg --url $producer.url login --verbose alice | from json +let bob = tg --url $producer.url login --verbose bob | from json + +# Alice has a local server that pushes to the producer as herself. +let alice_local = spawn --name alice-local --config { + remotes: { default: { url: $producer.url, token: $alice.token } }, +} + +# Alice creates a directory with a nested directory and pushes it to the producer. +let dir = tg --url $alice_local.url put 'tg.directory({ "sub": tg.directory({ "a.txt": tg.file("aaa"), "b.txt": tg.file("bbb") }) })' | str trim +let nested = tg --url $alice_local.url children $dir | from json | get 0 +let pushed = tg --url $alice_local.url --no-quiet push $dir | complete +success $pushed "Alice's push should succeed." + +# Alice grants Bob the nested directory subtree. +tg --url $producer.url --token $alice.token grant $bob.user.id object_subtree $nested | ignore + +# Wait for the detached indexing to settle, then kill the producer; the grant survives in the durable outbox. +sleep 5sec +let pid = open ($producer.directory | path join 'lock') | into int +kill --signal 2 $pid +if $nu.os-info.name == "linux" { + ^tail --pid $pid -f /dev/null +} else { + while (ps | where pid == $pid | is-not-empty) { sleep 10ms } +} + +# Start a new server on the same path with the indexer enabled to drain the outbox. +let indexer = spawn --name indexer --directory $directory --config { + authentication: { providers: { insecure: true } }, + indexer: { outbox: { partition_start: 0, partition_end: 256 } }, +} + +# Bob has a local server that talks to the indexer server as himself. +let bob_local = spawn --name bob-local --config { + remotes: { default: { url: $indexer.url, token: $bob.token } }, +} + +# Bob was granted the nested subtree and pulls it after the indexer drains the outbox. +let allowed = tg --url $bob_local.url --no-quiet pull $nested | complete +success $allowed "Bob should pull the granted nested directory subtree after the indexer drains the outbox." diff --git a/packages/cli/tests/grants/pull_survives_interrupted_indexing.nu b/packages/cli/tests/grants/pull_survives_interrupted_indexing.nu new file mode 100644 index 0000000000..3530871635 --- /dev/null +++ b/packages/cli/tests/grants/pull_survives_interrupted_indexing.nu @@ -0,0 +1,53 @@ +use ../../test.nu * + +# A grant that a push enqueues to the outbox survives the server being killed and authorizes a later pull once a second server with the indexer drains it. + +# The remote directory is shared by the producer and indexer phases. +let directory = mktemp -d + +# Start the producer with the indexer disabled so the push only enqueues the root grant to the outbox. +let producer = spawn --name producer --directory $directory --config { + authentication: { providers: { insecure: true } }, + indexer: { outbox: { partition_start: 0, partition_end: 0 } }, +} + +let alice = tg --url $producer.url login --verbose alice | from json +let bob = tg --url $producer.url login --verbose bob | from json + +# Alice has a local server that pushes to the producer as herself. +let alice_local = spawn --name alice-local --config { + remotes: { default: { url: $producer.url, token: $alice.token } }, +} + +# Alice creates a directory with a non-trivial subtree and pushes it to the producer. +let dir = tg --url $alice_local.url put 'tg.directory({ "a.txt": tg.file("aaa"), "b.txt": tg.file("bbb") })' | str trim +let pushed = tg --url $alice_local.url --no-quiet push $dir | complete +success $pushed "Alice's push should succeed." + +# Alice grants Bob the directory subtree. +tg --url $producer.url --token $alice.token grant $bob.user.id object_subtree $dir | ignore + +# Wait for the detached indexing to settle, then kill the producer; the grant survives in the durable outbox. +sleep 5sec +let pid = open ($producer.directory | path join 'lock') | into int +kill --signal 2 $pid +if $nu.os-info.name == "linux" { + ^tail --pid $pid -f /dev/null +} else { + while (ps | where pid == $pid | is-not-empty) { sleep 10ms } +} + +# Start a new server on the same path with the indexer enabled to drain the outbox. +let indexer = spawn --name indexer --directory $directory --config { + authentication: { providers: { insecure: true } }, + indexer: { outbox: { partition_start: 0, partition_end: 256 } }, +} + +# Bob has a local server that talks to the indexer server as himself. +let bob_local = spawn --name bob-local --config { + remotes: { default: { url: $indexer.url, token: $bob.token } }, +} + +# Bob was granted the subtree and pulls it after the indexer drains the outbox. +let allowed = tg --url $bob_local.url --no-quiet pull $dir | complete +success $allowed "Bob should pull the granted directory subtree after the indexer drains the outbox." diff --git a/packages/cli/tests/index/outbox.nu b/packages/cli/tests/index/outbox.nu new file mode 100644 index 0000000000..29f26144d5 --- /dev/null +++ b/packages/cli/tests/index/outbox.nu @@ -0,0 +1,34 @@ +use ../../test.nu * + +# Objects put on a server with no indexer are durably enqueued to the outbox, survive the server being killed, and are indexed by a second server started on the same path with the indexer enabled. + +# Create the directory shared by both servers. +let directory = mktemp -d + +# Spawn the first server with an indexer that has zero outbox workers, so it enqueues to the outbox without consuming it. +let producer = spawn --name producer --directory $directory --config { + indexer: { outbox: { partition_start: 0, partition_end: 0 } }, +} + +# Put an object tree. +let id = tg --url $producer.url put 'tg.directory({ "a.txt": tg.file("aaa"), "b.txt": tg.file("bbb") })' | str trim + +# Wait for the producer's detached outbox enqueue to commit, then kill the server. The producer never indexes the objects itself; it only enqueues them. +sleep 5sec +let pid = open ($producer.directory | path join 'lock') | into int +kill --signal 2 $pid +if $nu.os-info.name == "linux" { + ^tail --pid $pid -f /dev/null +} else { + while (ps | where pid == $pid | is-not-empty) { sleep 10ms } +} + +# Spawn the second server on the same path with an indexer that has outbox workers, so it drains the outbox. +let indexer = spawn --name indexer --directory $directory --config { + indexer: { outbox: { partition_start: 0, partition_end: 256 } }, +} + +# Wait for the outbox to drain and the index to propagate, then verify the metadata. +tg --url $indexer.url index +let metadata = tg --url $indexer.url object metadata $id | from json +assert equal $metadata.subtree.count 5 "the object tree should be indexed by the second server" diff --git a/packages/index/src/batch.rs b/packages/index/src/batch.rs index a0bb213854..d2331db619 100644 --- a/packages/index/src/batch.rs +++ b/packages/index/src/batch.rs @@ -1,24 +1,42 @@ use tangram_client::prelude::*; -#[derive(Clone, Debug, Default)] +#[derive(Clone, Debug, Default, serde::Deserialize, serde::Serialize)] pub struct Arg { + #[serde(default, skip_serializing_if = "Vec::is_empty")] pub delete_grants: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] pub delete_group_members: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] pub delete_groups: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] pub delete_organization_members: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] pub delete_organizations: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] pub delete_tags: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] pub delete_users: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] pub put_cache_entries: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] pub put_grants: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] pub put_group_members: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] pub put_groups: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] pub put_objects: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] pub put_organization_members: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] pub put_organizations: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] pub put_processes: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] pub put_sandboxes: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] pub put_tags: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] pub put_users: Vec, } diff --git a/packages/index/src/cache/put.rs b/packages/index/src/cache/put.rs index aeceec31e9..38ce4e52a7 100644 --- a/packages/index/src/cache/put.rs +++ b/packages/index/src/cache/put.rs @@ -1,6 +1,6 @@ use tangram_client::prelude::*; -#[derive(Clone, Debug)] +#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)] pub struct Arg { pub dependencies: Vec, pub id: tg::artifact::Id, diff --git a/packages/index/src/grant/delete.rs b/packages/index/src/grant/delete.rs index bedd2391cc..2968745a4d 100644 --- a/packages/index/src/grant/delete.rs +++ b/packages/index/src/grant/delete.rs @@ -1,6 +1,6 @@ use tangram_client::prelude::*; -#[derive(Clone, Debug)] +#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)] pub struct Arg { pub creator: Option, pub expires_at: Option, diff --git a/packages/index/src/grant/put.rs b/packages/index/src/grant/put.rs index cb98fa21d3..a2a27cd3b8 100644 --- a/packages/index/src/grant/put.rs +++ b/packages/index/src/grant/put.rs @@ -1,6 +1,6 @@ use tangram_client::prelude::*; -#[derive(Clone, Debug)] +#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)] pub struct Arg { pub created_at: i64, pub creator: Option, diff --git a/packages/index/src/group/member/delete.rs b/packages/index/src/group/member/delete.rs index 6c6d1d2e62..8b48a95c4c 100644 --- a/packages/index/src/group/member/delete.rs +++ b/packages/index/src/group/member/delete.rs @@ -1,6 +1,6 @@ use tangram_client::prelude::*; -#[derive(Clone, Debug)] +#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)] pub struct Arg { pub group: tg::group::Id, pub member: tg::group::Member, diff --git a/packages/index/src/group/member/put.rs b/packages/index/src/group/member/put.rs index 6c6d1d2e62..8b48a95c4c 100644 --- a/packages/index/src/group/member/put.rs +++ b/packages/index/src/group/member/put.rs @@ -1,6 +1,6 @@ use tangram_client::prelude::*; -#[derive(Clone, Debug)] +#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)] pub struct Arg { pub group: tg::group::Id, pub member: tg::group::Member, diff --git a/packages/index/src/group/put.rs b/packages/index/src/group/put.rs index 781217f182..f341ec39fb 100644 --- a/packages/index/src/group/put.rs +++ b/packages/index/src/group/put.rs @@ -1,6 +1,6 @@ use tangram_client::prelude::*; -#[derive(Clone, Debug)] +#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)] pub struct Arg { pub id: tg::group::Id, pub parent: Option, diff --git a/packages/index/src/object/put.rs b/packages/index/src/object/put.rs index d448d6dbdd..41e584ca99 100644 --- a/packages/index/src/object/put.rs +++ b/packages/index/src/object/put.rs @@ -1,6 +1,6 @@ use {super::Stored, std::collections::BTreeSet, tangram_client::prelude::*}; -#[derive(Clone, Debug)] +#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)] pub struct Arg { pub cache_entry: Option, pub children: BTreeSet, diff --git a/packages/index/src/organization/member/delete.rs b/packages/index/src/organization/member/delete.rs index aac20a9607..907478ca2d 100644 --- a/packages/index/src/organization/member/delete.rs +++ b/packages/index/src/organization/member/delete.rs @@ -1,6 +1,6 @@ use tangram_client::prelude::*; -#[derive(Clone, Debug)] +#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)] pub struct Arg { pub member: tg::organization::Member, pub organization: tg::organization::Id, diff --git a/packages/index/src/organization/member/put.rs b/packages/index/src/organization/member/put.rs index aac20a9607..907478ca2d 100644 --- a/packages/index/src/organization/member/put.rs +++ b/packages/index/src/organization/member/put.rs @@ -1,6 +1,6 @@ use tangram_client::prelude::*; -#[derive(Clone, Debug)] +#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)] pub struct Arg { pub member: tg::organization::Member, pub organization: tg::organization::Id, diff --git a/packages/index/src/organization/put.rs b/packages/index/src/organization/put.rs index f843432afd..26fd5cdd8a 100644 --- a/packages/index/src/organization/put.rs +++ b/packages/index/src/organization/put.rs @@ -1,6 +1,6 @@ use tangram_client::prelude::*; -#[derive(Clone, Debug)] +#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)] pub struct Arg { pub id: tg::organization::Id, pub specifier: tg::Specifier, diff --git a/packages/index/src/process/put.rs b/packages/index/src/process/put.rs index 51244ab22f..5a43c5225d 100644 --- a/packages/index/src/process/put.rs +++ b/packages/index/src/process/put.rs @@ -1,6 +1,6 @@ use {super::Stored, tangram_client::prelude::*}; -#[derive(Clone, Debug)] +#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)] pub struct Arg { pub children: Option>, pub command: tg::object::Id, diff --git a/packages/index/src/sandbox/put.rs b/packages/index/src/sandbox/put.rs index 0432ff2ff5..87af7025e0 100644 --- a/packages/index/src/sandbox/put.rs +++ b/packages/index/src/sandbox/put.rs @@ -1,6 +1,6 @@ use tangram_client::prelude::*; -#[derive(Clone, Debug)] +#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)] pub struct Arg { pub id: tg::sandbox::Id, pub owner: Option, diff --git a/packages/index/src/tag/put.rs b/packages/index/src/tag/put.rs index 8523a70525..ee6970c588 100644 --- a/packages/index/src/tag/put.rs +++ b/packages/index/src/tag/put.rs @@ -1,6 +1,6 @@ use tangram_client::prelude::*; -#[derive(Clone, Debug)] +#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)] pub struct Arg { pub id: tg::tag::Id, pub item: tg::Either, diff --git a/packages/index/src/user/put.rs b/packages/index/src/user/put.rs index 2e9f88c63f..b0200aa12c 100644 --- a/packages/index/src/user/put.rs +++ b/packages/index/src/user/put.rs @@ -1,6 +1,6 @@ use tangram_client::prelude::*; -#[derive(Clone, Debug)] +#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)] pub struct Arg { pub id: tg::user::Id, pub specifier: tg::Specifier, diff --git a/packages/server/src/checkin/artifact.rs b/packages/server/src/checkin/artifact.rs index 841021f8d3..635cdee4b2 100644 --- a/packages/server/src/checkin/artifact.rs +++ b/packages/server/src/checkin/artifact.rs @@ -15,7 +15,6 @@ use { path::Path, }, tangram_client::prelude::*, - tangram_index::prelude::*, tangram_object_store::prelude::*, }; @@ -919,22 +918,9 @@ impl Session { touched_at, }; self.server - .index_tasks - .spawn({ - let session = self.clone(); - move |_| async move { - let result = session - .server - .index - .batch(tangram_index::batch::Arg { - put_objects: vec![put_object_arg], - ..Default::default() - }) - .await; - if let Err(error) = result { - tracing::error!(error = %error.trace(), "failed to index the object"); - } - } + .index_objects_task(tangram_index::batch::Arg { + put_objects: vec![put_object_arg], + ..Default::default() }) .detach(); diff --git a/packages/server/src/checkin/index.rs b/packages/server/src/checkin/index.rs index 3f1fcd0129..daecde46f0 100644 --- a/packages/server/src/checkin/index.rs +++ b/packages/server/src/checkin/index.rs @@ -6,7 +6,6 @@ use { num::ToPrimitive as _, std::path::Path, tangram_client::prelude::*, - tangram_index::prelude::*, }; impl Session { @@ -86,15 +85,13 @@ impl Session { // Index. self.server - .index - .batch(tangram_index::batch::Arg { + .index_objects_task(tangram_index::batch::Arg { put_cache_entries: put_index_cache_entry_args, put_grants: put_grant.map(|arg| vec![arg]).unwrap_or_default(), put_objects: put_index_object_args, ..Default::default() }) - .await - .map_err(|error| tg::error!(!error, "failed to index"))?; + .await?; Ok(()) } diff --git a/packages/server/src/config.rs b/packages/server/src/config.rs index 87482dbefd..58b5db3acb 100644 --- a/packages/server/src/config.rs +++ b/packages/server/src/config.rs @@ -419,6 +419,7 @@ pub struct LmdbIndex { pub path: PathBuf, } +#[serde_as] #[derive(Clone, Debug, serde::Deserialize, serde::Serialize)] #[serde(default, deny_unknown_fields)] pub struct Indexer { @@ -426,11 +427,26 @@ pub struct Indexer { pub concurrency: usize, + #[serde_as(as = "BoolOptionDefault")] + pub outbox: Option, + pub partition_count: u64, pub partition_start: u64, } +#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)] +#[serde(default, deny_unknown_fields)] +pub struct Outbox { + pub batch_size: usize, + + pub max_partitions: u64, + + pub partition_end: u64, + + pub partition_start: u64, +} + #[derive(Clone, Debug, Default, serde::Deserialize, serde::Serialize)] #[serde(default, deny_unknown_fields)] pub struct Logs { @@ -1248,12 +1264,24 @@ impl Default for Indexer { Self { batch_size: 1024, concurrency: 1, + outbox: Some(Outbox::default()), partition_count: 256, partition_start: 0, } } } +impl Default for Outbox { + fn default() -> Self { + Self { + batch_size: 1024, + max_partitions: 256, + partition_end: 256, + partition_start: 0, + } + } +} + impl Default for LogStore { fn default() -> Self { Self::Lmdb(LmdbLogStore::default()) diff --git a/packages/server/src/index.rs b/packages/server/src/index.rs index 89d115b904..157c43ff0a 100644 --- a/packages/server/src/index.rs +++ b/packages/server/src/index.rs @@ -466,16 +466,42 @@ impl Session { } } - // Wait for remote object put tasks to enqueue their index tasks. + // Wait for the remote object put tasks to enqueue their index tasks, then for the index tasks to finish. The index tasks enqueue the object outbox entries, so they must finish before the outbox is snapshotted below. progress.spinner("tasks", "waiting for tasks"); self.server.remote_object_put_tasks.wait().await; - progress.finish("tasks"); - - // Wait for outstanding index tasks to finish. - progress.spinner("tasks", "waiting for tasks"); self.server.index_tasks.wait().await; progress.finish("tasks"); + // Snapshot the max seq of the object outbox and wait for all partitions to drain up to it. This only applies to servers that run the outbox consumer. + if let Some(config) = self + .server + .config + .indexer + .as_ref() + .and_then(|indexer| indexer.outbox.as_ref()) + { + let partitions = (0..config.max_partitions) + .map(|partition| partition.to_i64().unwrap()) + .collect::>(); + if let Some(snapshot) = self + .server + .object_store + .object_outbox_max_seq(partitions.clone()) + .await? + { + progress.spinner("outbox", "waiting for the object outbox"); + while self + .server + .object_store + .object_outbox_has_pending(partitions.clone(), snapshot) + .await? + { + tokio::time::sleep(Duration::from_millis(100)).await; + } + progress.finish("outbox"); + } + } + // Subscribe to indexer progress. let wakeups = self .server @@ -513,7 +539,118 @@ impl Session { } impl Server { + pub(crate) fn index_objects( + &self, + arg: tangram_index::batch::Arg, + ) -> tangram_futures::task::Shared<()> { + self.index_tasks.spawn({ + let server = self.clone(); + |_| async move { + server + .index_objects_task(arg) + .await + .inspect_err(|error| { + tracing::error!(error = %error.trace(), "failed to index the objects"); + }) + .ok(); + } + }) + } + + pub(crate) async fn index_objects_task( + &self, + mut arg: tangram_index::batch::Arg, + ) -> tg::Result<()> { + // Index directly if there is no outbox. + let Some(outbox) = self + .config + .indexer + .as_ref() + .and_then(|indexer| indexer.outbox.clone()) + .filter(|_| self.object_store.supports_outbox()) + else { + return self.index.batch(arg).await; + }; + + let put_objects = std::mem::take(&mut arg.put_objects); + let object_ids = put_objects + .iter() + .map(|object| tg::Id::from(object.id.clone())) + .collect::>(); + let mut grants: std::collections::HashMap<_, Vec<_>> = std::collections::HashMap::new(); + let mut other_grants = Vec::new(); + for grant in std::mem::take(&mut arg.put_grants) { + if object_ids.contains(&grant.resource) { + grants.entry(grant.resource.clone()).or_default().push(grant); + } else { + other_grants.push(grant); + } + } + arg.put_grants = other_grants; + + // Build the outbox entries. Each object is enqueued as a row whose payload is a serialized batch arg carrying its put and, when present, its grant. The batch is written to a single random partition, which only distributes load since entries are never located by id. + let partition = rand::random_range(0..outbox.max_partitions) + .to_i64() + .unwrap(); + let mut entries = Vec::with_capacity(put_objects.len()); + for object in put_objects { + let object_grants = grants + .remove(&tg::Id::from(object.id.clone())) + .unwrap_or_default(); + let arg = index::batch::Arg { + put_grants: object_grants, + put_objects: vec![object], + ..Default::default() + }; + let payload = serde_json::to_vec(&arg) + .map_err(|error| tg::error!(!error, "failed to serialize the outbox payload"))?; + entries.push((partition, payload)); + } + + // Enqueue the objects to the outbox. + self.enqueue_object_outbox(entries).await?; + + // Index any remaining non-object args directly. + if !arg.is_empty() { + self.index.batch(arg).await?; + } + + Ok(()) + } + + // Enqueue a batch to the object outbox, retrying transient failures with exponential backoff so a durable index intent is not lost to a momentary store error. + pub(crate) async fn enqueue_object_outbox( + &self, + entries: Vec<(i64, Vec)>, + ) -> tg::Result<()> { + let options = tangram_futures::retry::Options::default(); + tangram_futures::retry::retry(&options, || async { + match self.object_store.enqueue_object_outbox(entries.clone()).await { + Ok(()) => Ok(std::ops::ControlFlow::Break(())), + Err(error) => Ok(std::ops::ControlFlow::Continue(error)), + } + }) + .await + } + pub(crate) async fn indexer_task(&self, config: &crate::config::Indexer) -> tg::Result<()> { + // Drain the object outbox in a task for stores that support it. The task is aborted when the indexer task is dropped. + let _object_outbox_task = config + .outbox + .clone() + .filter(|_| self.object_store.supports_outbox()) + .map(|config| { + Task::spawn({ + let server = self.clone(); + |_| async move { + let result = server.object_outbox_task(&config).await; + if let Err(error) = result { + tracing::error!(error = %error.trace(), "the object outbox task failed"); + } + } + }) + }); + let partition_start = config.partition_start; let partition_count = config.partition_count; let concurrency = config.concurrency.to_u64().unwrap(); @@ -548,6 +685,105 @@ impl Server { } } } + + pub(crate) async fn object_outbox_task( + &self, + config: &crate::config::Outbox, + ) -> tg::Result<()> { + // Run one worker per partition in this node's assigned range. The range is empty for nodes that only enqueue, which then consume nothing. + let futures = (config.partition_start..config.partition_end).map(|partition| { + self.object_outbox_worker(config.batch_size, vec![partition.to_i64().unwrap()]) + }); + future::try_join_all(futures).await?; + Ok(()) + } + + async fn object_outbox_worker( + &self, + batch_size: usize, + partitions: Vec, + ) -> tg::Result<()> { + loop { + match self + .drain_object_outbox(batch_size, partitions.clone()) + .await + { + Ok(0) => { + tokio::time::sleep(Duration::from_millis(100)).await; + }, + Ok(_) => { + self.messenger + .publish("indexer_progress".to_owned(), ()) + .await + .ok(); + }, + Err(error) => { + tracing::error!(error = %error.trace(), "failed to drain the object outbox"); + tokio::time::sleep(Duration::from_secs(1)).await; + }, + } + } + } + + async fn drain_object_outbox( + &self, + batch_size: usize, + partitions: Vec, + ) -> tg::Result { + let items = self + .object_store + .dequeue_object_outbox(partitions, batch_size.to_i32().unwrap()) + .await?; + if items.is_empty() { + return Ok(0); + } + + let mut arg = index::batch::Arg::default(); + let mut seqs = Vec::with_capacity(items.len()); + for item in items { + seqs.push((item.partition, item.seq)); + let entry = serde_json::from_slice::(&item.payload) + .map_err(|error| tg::error!(!error, "failed to deserialize the outbox payload"))?; + arg.put_grants.extend(entry.put_grants); + arg.put_objects.extend(entry.put_objects); + } + + if !arg.is_empty() { + self.index + .batch(arg) + .await + .map_err(|error| tg::error!(!error, "failed to put the objects to the index"))?; + } + + self.object_store.delete_object_outbox(&seqs).await?; + + Ok(seqs.len()) + } + + pub(crate) fn index_processes( + &self, + arg: tangram_index::batch::Arg, + ) -> tangram_futures::task::Shared<()> { + self.index_tasks.spawn({ + let server = self.clone(); + |_| async move { + server + .index_process_task(arg) + .await + .inspect_err(|error| { + tracing::error!(error = %error.trace(), "failed to index the processes"); + }) + .ok(); + } + }) + } + + pub(crate) async fn index_process_task( + &self, + arg: tangram_index::batch::Arg, + ) -> tg::Result<()> { + self.index.batch(arg).await + } } impl Session { diff --git a/packages/server/src/object/batch.rs b/packages/server/src/object/batch.rs index 00028bd636..cc8c459dab 100644 --- a/packages/server/src/object/batch.rs +++ b/packages/server/src/object/batch.rs @@ -6,7 +6,6 @@ use { tangram_http::{ body::Boxed as BoxBody, request::Ext as _, response::Ext as _, response::builder::Ext as _, }, - tangram_index::prelude::*, tangram_object_store::prelude::*, }; @@ -52,7 +51,7 @@ impl Session { .to_i64() .unwrap(); - // Store the objects. + // Build the store put args. let principal = (!matches!(self.context.principal, tg::Principal::Anonymous)) .then(|| self.context.principal.clone()); let put_args: Vec<_> = arg @@ -65,11 +64,6 @@ impl Session { stored_at: now, }) .collect(); - self.server - .object_store - .put_batch(put_args) - .await - .map_err(|error| tg::error!(!error, "failed to put the objects"))?; // Deserialize the objects and create the index args. let mut batch_objects = BTreeSet::new(); @@ -136,44 +130,46 @@ impl Session { } } - let mut put_grant_args = Vec::with_capacity(arg.objects.len()); - for object in &arg.objects { - if let Some(principal) = &principal { - let permission = if subtree_objects.contains(&object.id) { - tg::grant::permission::object::Permission::Subtree - } else { - tg::grant::permission::object::Permission::Node - }; - put_grant_args.push(tangram_index::grant::put::Arg { - created_at: now, - creator: Some(principal.clone()), - expires_at: Some(grant_expires_at), - permissions: tg::grant::Permission::Object(permission).into(), - principal: principal.try_to_grant_principal()?, - resource: object.id.clone().into(), - }); - } - } + // Compute the grant for each object. + let grants = arg + .objects + .iter() + .map(|object| { + principal + .as_ref() + .map(|principal| { + let permission = if subtree_objects.contains(&object.id) { + tg::grant::permission::object::Permission::Subtree + } else { + tg::grant::permission::object::Permission::Node + }; + Ok::<_, tg::Error>(tangram_index::grant::put::Arg { + created_at: now, + creator: Some(principal.clone()), + expires_at: Some(grant_expires_at), + permissions: tg::grant::Permission::Object(permission).into(), + principal: principal.try_to_grant_principal()?, + resource: object.id.clone().into(), + }) + }) + .transpose() + }) + .collect::>>()?; - // Spawn a task to index the objects. + // Put the objects. self.server - .index_tasks - .spawn(|_| { - let session = self.clone(); - async move { - if let Err(error) = session - .server - .index - .batch(tangram_index::batch::Arg { - put_grants: put_grant_args, - put_objects: put_object_args, - ..Default::default() - }) - .await - { - tracing::error!(error = %error.trace(), "failed to put object batch to index"); - } - } + .object_store + .put_batch(put_args) + .await + .map_err(|error| tg::error!(!error, "failed to put the objects"))?; + + // Index the objects. + let put_grant_args = grants.into_iter().flatten().collect::>(); + self.server + .index_objects(tangram_index::batch::Arg { + put_grants: put_grant_args, + put_objects: put_object_args, + ..Default::default() }) .detach(); diff --git a/packages/server/src/object/put.rs b/packages/server/src/object/put.rs index faaa188800..c386605abe 100644 --- a/packages/server/src/object/put.rs +++ b/packages/server/src/object/put.rs @@ -6,7 +6,6 @@ use { tangram_http::{ body::Boxed as BoxBody, request::Ext as _, response::Ext as _, response::builder::Ext as _, }, - tangram_index::prelude::*, tangram_object_store::prelude::*, }; @@ -56,11 +55,6 @@ impl Session { id: id.clone(), stored_at: now, }; - self.server - .object_store - .put(put_arg) - .await - .map_err(|error| tg::error!(!error, "failed to put the object"))?; let data = tg::object::Data::deserialize(id.kind(), arg.bytes.clone()) .map_err(|error| tg::error!(!error, "failed to deserialize the object"))?; @@ -77,46 +71,6 @@ impl Session { tg::grant::permission::object::Permission::Node }; - let (node_solvable, node_solved) = match data { - tg::object::Data::File(file) => match file { - tg::file::Data::Pointer(_) => (false, true), - tg::file::Data::Node(node) => (node.solvable(), node.solved()), - }, - tg::object::Data::Graph(graph) => { - graph - .nodes - .iter() - .fold((false, true), |(solvable, solved), node| { - if let tg::graph::data::Node::File(file) = node { - (solvable || file.solvable(), solved && file.solved()) - } else { - (solvable, solved) - } - }) - }, - _ => (false, true), - }; - - let metadata = if let Some(metadata) = arg.metadata { - metadata - } else { - tg::object::Metadata { - node: tg::object::metadata::Node { - size: arg.bytes.len().to_u64().unwrap(), - solvable: node_solvable, - solved: node_solved, - }, - ..Default::default() - } - }; - let arg = tangram_index::object::put::Arg { - cache_entry: None, - children, - id: id.clone(), - metadata, - stored: tangram_index::object::Stored::default(), - touched_at: now, - }; let put_grant = (!matches!(self.context.principal, tg::Principal::Anonymous)) .then(|| { let principal = self.context.principal.clone(); @@ -130,21 +84,20 @@ impl Session { }) }) .transpose()?; + self.server - .index_tasks - .spawn(|_| { - let session = self.clone(); - async move { - let arg = tangram_index::batch::Arg { - put_grants: put_grant.map(|arg| vec![arg]).unwrap_or_default(), - put_objects: vec![arg], - ..Default::default() - }; - let result = session.server.index.batch(arg).await; - if let Err(error) = result { - tracing::error!(error = %error.trace(), "failed to put the object to the index"); - } - } + .object_store + .put(put_arg) + .await + .map_err(|error| tg::error!(!error, "failed to put the object"))?; + + let size = arg.bytes.len().to_u64().unwrap(); + let put_object = object_index_put_arg(id.clone(), &data, children, size, arg.metadata, now); + self.server + .index_objects(tangram_index::batch::Arg { + put_grants: put_grant.map(|arg| vec![arg]).unwrap_or_default(), + put_objects: vec![put_object], + ..Default::default() }) .detach(); @@ -286,3 +239,26 @@ impl Session { Ok(response) } } + +// Compute the index put argument for an object from its data, recomputing the node metadata unless the metadata is overridden. +pub(crate) fn object_index_put_arg( + id: tg::object::Id, + data: &tg::object::Data, + children: BTreeSet, + size: u64, + metadata: Option, + touched_at: i64, +) -> tangram_index::object::put::Arg { + let metadata = metadata.unwrap_or_else(|| tg::object::Metadata { + node: tg::object::metadata::Node::with_data_and_size(data, size), + ..Default::default() + }); + tangram_index::object::put::Arg { + cache_entry: None, + children, + id, + metadata, + stored: tangram_index::object::Stored::default(), + touched_at, + } +} diff --git a/packages/server/src/object/store.rs b/packages/server/src/object/store.rs index 587ae35da4..ac58cabc13 100644 --- a/packages/server/src/object/store.rs +++ b/packages/server/src/object/store.rs @@ -1,6 +1,6 @@ #[cfg(feature = "lmdb")] use std::path::Path; -use {tangram_client::prelude::*, tangram_object_store as object_store}; +use {bytes::Bytes, tangram_client::prelude::*, tangram_object_store as object_store}; pub use object_store::{CachePointer, DeleteArg, PutArg, TryGetArg, TryGetBatchArg, TryGetOutput}; @@ -17,6 +17,12 @@ pub enum Store { Scylla(object_store::scylla::Store), } +pub struct OutboxItem { + pub seq: uuid::Uuid, + pub partition: i64, + pub payload: Bytes, +} + impl Store { #[cfg(feature = "lmdb")] pub fn new_lmdb(directory: &Path, config: &crate::config::LmdbObjectStore) -> tg::Result { @@ -197,6 +203,139 @@ impl Store { } Ok(()) } + + pub fn supports_outbox(&self) -> bool { + match self { + #[cfg(feature = "lmdb")] + Self::Lmdb(_) => true, + Self::Memory(_) => true, + #[cfg(feature = "scylla")] + Self::Scylla(_) => true, + } + } + + pub async fn enqueue_object_outbox(&self, entries: Vec<(i64, Vec)>) -> tg::Result<()> { + match self { + #[cfg(feature = "lmdb")] + Self::Lmdb(lmdb) => lmdb.enqueue_object_outbox(entries).await, + Self::Memory(memory) => { + memory.enqueue_object_outbox(entries); + Ok(()) + }, + #[cfg(feature = "scylla")] + Self::Scylla(scylla) => scylla.enqueue_object_outbox(entries).await, + } + } + + pub async fn dequeue_object_outbox( + &self, + partitions: Vec, + limit: i32, + ) -> tg::Result> { + match self { + #[cfg(feature = "lmdb")] + Self::Lmdb(lmdb) => { + let items = lmdb + .dequeue_object_outbox(partitions, limit) + .await? + .into_iter() + .map(|item| OutboxItem { + seq: uuid::Uuid::from_bytes(item.seq), + partition: item.partition, + payload: item.payload, + }) + .collect(); + Ok(items) + }, + Self::Memory(memory) => { + let items = memory + .dequeue_object_outbox(&partitions, limit) + .into_iter() + .map(|item| OutboxItem { + seq: uuid::Uuid::from_bytes(item.seq), + partition: item.partition, + payload: item.payload, + }) + .collect(); + Ok(items) + }, + #[cfg(feature = "scylla")] + Self::Scylla(scylla) => { + let items = scylla + .dequeue_object_outbox(partitions, limit) + .await? + .into_iter() + .map(|item| OutboxItem { + seq: uuid::Uuid::from_bytes(item.seq), + partition: item.partition, + payload: item.payload, + }) + .collect(); + Ok(items) + }, + } + } + + pub async fn delete_object_outbox(&self, entries: &[(i64, uuid::Uuid)]) -> tg::Result<()> { + let entries = entries + .iter() + .map(|(partition, token)| (*partition, *token.as_bytes())) + .collect::>(); + match self { + #[cfg(feature = "lmdb")] + Self::Lmdb(lmdb) => lmdb.delete_object_outbox(&entries).await, + Self::Memory(memory) => { + memory.delete_object_outbox(&entries); + Ok(()) + }, + #[cfg(feature = "scylla")] + Self::Scylla(scylla) => scylla.delete_object_outbox(&entries).await, + } + } + + pub async fn object_outbox_max_seq( + &self, + partitions: Vec, + ) -> tg::Result> { + match self { + #[cfg(feature = "lmdb")] + Self::Lmdb(lmdb) => Ok(lmdb + .object_outbox_max_seq(partitions) + .await? + .map(uuid::Uuid::from_bytes)), + Self::Memory(memory) => Ok(memory + .object_outbox_max_seq(&partitions) + .map(uuid::Uuid::from_bytes)), + #[cfg(feature = "scylla")] + Self::Scylla(scylla) => Ok(scylla + .object_outbox_max_seq(partitions) + .await? + .map(uuid::Uuid::from_bytes)), + } + } + + pub async fn object_outbox_has_pending( + &self, + partitions: Vec, + seq: uuid::Uuid, + ) -> tg::Result { + match self { + #[cfg(feature = "lmdb")] + Self::Lmdb(lmdb) => { + lmdb.object_outbox_has_pending(partitions, *seq.as_bytes()) + .await + }, + Self::Memory(memory) => { + Ok(memory.object_outbox_has_pending(&partitions, *seq.as_bytes())) + }, + #[cfg(feature = "scylla")] + Self::Scylla(scylla) => { + scylla + .object_outbox_has_pending(partitions, *seq.as_bytes()) + .await + }, + } + } } impl object_store::Store for Store { diff --git a/packages/server/src/process/finalize.rs b/packages/server/src/process/finalize.rs index 7508820bb9..5ee7c297b9 100644 --- a/packages/server/src/process/finalize.rs +++ b/packages/server/src/process/finalize.rs @@ -6,7 +6,6 @@ use { tangram_client::prelude::*, tangram_database::{self as db, prelude::*}, tangram_futures::{stream::Ext as _, task::Stopper}, - tangram_index::prelude::*, tangram_messenger::prelude::*, tokio_stream::wrappers::IntervalStream, }; @@ -293,23 +292,11 @@ impl Server { stored: tangram_index::process::Stored::default(), touched_at: now, }; - self.index_tasks - .spawn(|_| { - let server = self.clone(); - async move { - if let Err(error) = server - .index - .batch(tangram_index::batch::Arg { - put_processes: vec![put_process_arg], - ..Default::default() - }) - .await - { - tracing::error!(error = %error.trace(), "failed to put process to index"); - } - } - }) - .detach(); + self.index_processes(tangram_index::batch::Arg { + put_processes: vec![put_process_arg], + ..Default::default() + }) + .detach(); Ok(()) } diff --git a/packages/server/src/process/finish.rs b/packages/server/src/process/finish.rs index 7bcf1a735b..ce95b0d515 100644 --- a/packages/server/src/process/finish.rs +++ b/packages/server/src/process/finish.rs @@ -11,7 +11,6 @@ use { tangram_http::{ body::Boxed as BoxBody, request::Ext as _, response::Ext as _, response::builder::Ext as _, }, - tangram_index::prelude::*, }; #[cfg(feature = "postgres")] @@ -243,8 +242,7 @@ impl Session { touched_at: time::OffsetDateTime::now_utc().unix_timestamp(), }; if let Err(error) = server - .index - .batch(tangram_index::batch::Arg { + .index_process_task(tangram_index::batch::Arg { put_processes: vec![put_process_arg], ..Default::default() }) diff --git a/packages/server/src/process/put.rs b/packages/server/src/process/put.rs index 7e367545ae..4524df50c4 100644 --- a/packages/server/src/process/put.rs +++ b/packages/server/src/process/put.rs @@ -6,7 +6,6 @@ use { tangram_http::{ body::Boxed as BoxBody, request::Ext as _, response::Ext as _, response::builder::Ext as _, }, - tangram_index::prelude::*, }; #[cfg(feature = "postgres")] @@ -158,23 +157,10 @@ impl Session { }) .transpose()?; self.server - .index_tasks - .spawn(|_| { - let session = self.clone(); - async move { - if let Err(error) = session - .server - .index - .batch(tangram_index::batch::Arg { - put_grants: put_grant.map(|arg| vec![arg]).unwrap_or_default(), - put_processes: vec![put_process_arg], - ..Default::default() - }) - .await - { - tracing::error!(error = %error.trace(), "failed to put process to index"); - } - } + .index_processes(tangram_index::batch::Arg { + put_grants: put_grant.map(|arg| vec![arg]).unwrap_or_default(), + put_processes: vec![put_process_arg], + ..Default::default() }) .detach(); diff --git a/packages/server/src/process/spawn.rs b/packages/server/src/process/spawn.rs index 5e851ae63b..f7bda1fb9e 100644 --- a/packages/server/src/process/spawn.rs +++ b/packages/server/src/process/spawn.rs @@ -250,18 +250,7 @@ impl Session { None }; if let Some(index_batch_arg) = index_batch_arg { - self.server - .index_tasks - .spawn(|_| { - let server = self.server.clone(); - async move { - let result = server.index.batch(index_batch_arg).await; - if let Err(error) = result { - tracing::error!(error = %error.trace(), "failed to put process to index"); - } - } - }) - .detach(); + self.server.index_processes(index_batch_arg).detach(); } // Wake the watchdog so depth-based limits are enforced promptly. @@ -2213,18 +2202,7 @@ impl Session { put_processes: vec![arg], ..Default::default() }; - self.server - .index_tasks - .spawn(|_| { - let server = self.server.clone(); - async move { - let result = server.index.batch(arg).await; - if let Err(error) = result { - tracing::error!(error = %error.trace(), "failed to put process to index"); - } - } - }) - .detach(); + self.server.index_processes(arg).detach(); } fn spawn_publish_process_child_message_task(&self, parent: &tg::process::Id) { diff --git a/packages/server/src/sync/get.rs b/packages/server/src/sync/get.rs index dd0ffc9812..df612a3d99 100644 --- a/packages/server/src/sync/get.rs +++ b/packages/server/src/sync/get.rs @@ -167,6 +167,9 @@ impl Session { // Await the futures. future::try_join4(input_future, queue_future, index_future, store_future).await?; + // Enqueue the root grants to the outbox before returning so authorization survives interrupted indexing. + self.sync_get_enqueue_root_grants(&state).await?; + // Stop and await the progress task. progress_task.stop(); progress_task diff --git a/packages/server/src/sync/get/index.rs b/packages/server/src/sync/get/index.rs index ff5d544709..ea006bb2e4 100644 --- a/packages/server/src/sync/get/index.rs +++ b/packages/server/src/sync/get/index.rs @@ -297,6 +297,69 @@ impl Session { Ok(()) } + pub(super) async fn sync_get_enqueue_root_grants(&self, state: &State) -> tg::Result<()> { + let Some(outbox) = self + .server + .config + .indexer + .as_ref() + .and_then(|indexer| indexer.outbox.as_ref()) + .filter(|_| self.server.object_store.supports_outbox()) + else { + return Ok(()); + }; + + let grant_principal = match &self.context.principal { + tg::Principal::Root => return Ok(()), + tg::Principal::Anonymous => tg::grant::Principal::Public, + principal => principal.try_to_grant_principal()?, + }; + + let created_at = time::OffsetDateTime::now_utc().unix_timestamp(); + let expires_at = created_at + + self + .server + .config + .object + .grant_time_to_live + .as_secs() + .to_i64() + .unwrap(); + + let partition = rand::random_range(0..outbox.max_partitions) + .to_i64() + .unwrap(); + let mut entries = Vec::new(); + for item in &state.arg.get { + let tg::Either::Left(id) = item else { + continue; + }; + let grant = tangram_index::grant::put::Arg { + created_at, + creator: Some(self.context.principal.clone()), + expires_at: Some(expires_at), + permissions: tg::grant::permission::Set::Object( + tg::grant::permission::object::Set::from_permission( + tg::grant::permission::object::Permission::Subtree, + ), + ), + principal: grant_principal.clone(), + resource: id.clone().into(), + }; + let arg = tangram_index::batch::Arg { + put_grants: vec![grant], + ..Default::default() + }; + let payload = serde_json::to_vec(&arg) + .map_err(|error| tg::error!(!error, "failed to serialize the grant"))?; + entries.push((partition, payload)); + } + + self.server.enqueue_object_outbox(entries).await?; + + Ok(()) + } + fn sync_get_index_create_args( &self, graph: &mut Graph, diff --git a/packages/server/src/write.rs b/packages/server/src/write.rs index 014513ce9b..9b4011459b 100644 --- a/packages/server/src/write.rs +++ b/packages/server/src/write.rs @@ -13,7 +13,6 @@ use { }, tangram_client::prelude::*, tangram_http::{body::Boxed as BoxBody, request::Ext as _}, - tangram_index::prelude::*, tangram_object_store::prelude::*, tokio::io::{AsyncRead, AsyncWriteExt as _}, }; @@ -504,24 +503,11 @@ impl Session { grant_expires_at, ); self.server - .index_tasks - .spawn(|_| { - let session = self.clone(); - async move { - if let Err(error) = session - .server - .index - .batch(tangram_index::batch::Arg { - put_cache_entries: put_cache_entry_args, - put_grants: put_grant_args, - put_objects: put_object_args, - ..Default::default() - }) - .await - { - tracing::error!(error = %error.trace(), "failed to index the write"); - } - } + .index_objects(tangram_index::batch::Arg { + put_cache_entries: put_cache_entry_args, + put_grants: put_grant_args, + put_objects: put_object_args, + ..Default::default() }) .detach(); Ok(()) diff --git a/packages/stores/object/src/lmdb.rs b/packages/stores/object/src/lmdb.rs index f23011d80a..3b8cd9ffbd 100644 --- a/packages/stores/object/src/lmdb.rs +++ b/packages/stores/object/src/lmdb.rs @@ -8,9 +8,12 @@ use { mod delete; mod flush; mod get; +mod outbox; mod put; mod task; +pub use outbox::Item; + #[derive(Clone, Debug)] pub struct Config { pub map_size: usize, @@ -21,6 +24,8 @@ pub struct Store { db: Db, env: lmdb::Env, handle: Option>, + object_outbox_db: Db, + seq: std::sync::atomic::AtomicU64, sender: Option, } @@ -63,7 +68,7 @@ impl Store { let env = unsafe { lmdb::EnvOpenOptions::new() .map_size(config.map_size) - .max_dbs(3) + .max_dbs(2) .max_readers(1_000) .flags( lmdb::EnvFlags::NO_SUB_DIR @@ -79,10 +84,21 @@ impl Store { let db = env .create_database(&mut transaction, None) .map_err(|error| tg::error!(!error, "failed to create the database"))?; + let object_outbox_db = env + .create_database(&mut transaction, Some("object_outbox")) + .map_err(|error| tg::error!(!error, "failed to create the object outbox database"))?; transaction .commit() .map_err(|error| tg::error!(!error, "failed to commit the transaction"))?; + // Initialize the sequence counter to the maximum sequence already persisted so new sequences are monotonic across restarts. + let transaction = env + .read_txn() + .map_err(|error| tg::error!(!error, "failed to begin a transaction"))?; + let seq = Self::outbox_last_seq(&object_outbox_db, &transaction)?; + drop(transaction); + let seq = std::sync::atomic::AtomicU64::new(seq); + // Create the thread. let (sender, receiver) = tokio::sync::mpsc::channel(256); let handle = std::thread::spawn({ @@ -94,6 +110,8 @@ impl Store { db, env, handle: Some(handle), + object_outbox_db, + seq, sender: Some(sender), }) } diff --git a/packages/stores/object/src/lmdb/outbox.rs b/packages/stores/object/src/lmdb/outbox.rs new file mode 100644 index 0000000000..6e6c7d7ee9 --- /dev/null +++ b/packages/stores/object/src/lmdb/outbox.rs @@ -0,0 +1,211 @@ +use { + super::{Db, Store}, + bytes::Bytes, + heed as lmdb, + num::ToPrimitive as _, + std::sync::atomic::Ordering, + tangram_client::prelude::*, +}; + +pub mod object; + +pub struct Item { + pub seq: [u8; 16], + pub partition: i64, + pub payload: Bytes, +} + +impl Store { + pub(super) async fn enqueue_outbox_inner( + &self, + db: Db, + entries: Vec<(i64, Vec)>, + ) -> tg::Result<()> { + if entries.is_empty() { + return Ok(()); + } + + // Assign a monotonic sequence to each entry. + let rows = entries + .into_iter() + .map(|(partition, payload)| { + let seq = self.seq.fetch_add(1, Ordering::Relaxed) + 1; + (seq_to_key(seq), encode_value(partition, &payload)) + }) + .collect::>(); + + let env = self.env.clone(); + tokio::task::spawn_blocking(move || { + let mut transaction = env + .write_txn() + .map_err(|error| tg::error!(!error, "failed to begin a transaction"))?; + for (key, value) in &rows { + db.put(&mut transaction, key, value) + .map_err(|error| tg::error!(!error, "failed to put the outbox entry"))?; + } + transaction + .commit() + .map_err(|error| tg::error!(!error, "failed to commit the transaction"))?; + Ok(()) + }) + .await + .map_err(|error| tg::error!(!error, "failed to join the task"))? + } + + pub(super) async fn dequeue_outbox_inner( + &self, + db: Db, + partitions: Vec, + limit: i32, + ) -> tg::Result> { + let limit = limit.to_usize().unwrap_or(0); + let env = self.env.clone(); + tokio::task::spawn_blocking(move || { + let transaction = env + .read_txn() + .map_err(|error| tg::error!(!error, "failed to begin a transaction"))?; + let mut items = Vec::new(); + for entry in db + .iter(&transaction) + .map_err(|error| tg::error!(!error, "failed to iterate the outbox"))? + { + if items.len() >= limit { + break; + } + let (key, value) = + entry.map_err(|error| tg::error!(!error, "failed to get the outbox entry"))?; + let partition = decode_partition(value)?; + if partitions.contains(&partition) { + items.push(Item { + seq: seq_from_key(key)?, + partition, + payload: Bytes::copy_from_slice(&value[8..]), + }); + } + } + Ok(items) + }) + .await + .map_err(|error| tg::error!(!error, "failed to join the task"))? + } + + pub(super) async fn delete_outbox_inner( + &self, + db: Db, + entries: &[(i64, [u8; 16])], + ) -> tg::Result<()> { + if entries.is_empty() { + return Ok(()); + } + let keys = entries.iter().map(|(_, seq)| *seq).collect::>(); + let env = self.env.clone(); + tokio::task::spawn_blocking(move || { + let mut transaction = env + .write_txn() + .map_err(|error| tg::error!(!error, "failed to begin a transaction"))?; + for key in &keys { + db.delete(&mut transaction, key) + .map_err(|error| tg::error!(!error, "failed to delete the outbox entry"))?; + } + transaction + .commit() + .map_err(|error| tg::error!(!error, "failed to commit the transaction"))?; + Ok(()) + }) + .await + .map_err(|error| tg::error!(!error, "failed to join the task"))? + } + + pub(super) async fn outbox_max_seq_inner( + &self, + db: Db, + partitions: Vec, + ) -> tg::Result> { + let env = self.env.clone(); + tokio::task::spawn_blocking(move || { + let transaction = env + .read_txn() + .map_err(|error| tg::error!(!error, "failed to begin a transaction"))?; + for entry in db + .rev_iter(&transaction) + .map_err(|error| tg::error!(!error, "failed to iterate the outbox"))? + { + let (key, value) = + entry.map_err(|error| tg::error!(!error, "failed to get the outbox entry"))?; + if partitions.contains(&decode_partition(value)?) { + return Ok(Some(seq_from_key(key)?)); + } + } + Ok(None) + }) + .await + .map_err(|error| tg::error!(!error, "failed to join the task"))? + } + + pub(super) async fn outbox_has_pending_inner( + &self, + db: Db, + partitions: Vec, + seq: [u8; 16], + ) -> tg::Result { + let env = self.env.clone(); + tokio::task::spawn_blocking(move || { + let transaction = env + .read_txn() + .map_err(|error| tg::error!(!error, "failed to begin a transaction"))?; + for entry in db + .iter(&transaction) + .map_err(|error| tg::error!(!error, "failed to iterate the outbox"))? + { + let (key, value) = + entry.map_err(|error| tg::error!(!error, "failed to get the outbox entry"))?; + if key > seq.as_slice() { + break; + } + if partitions.contains(&decode_partition(value)?) { + return Ok(true); + } + } + Ok(false) + }) + .await + .map_err(|error| tg::error!(!error, "failed to join the task"))? + } + + pub(super) fn outbox_last_seq(db: &Db, transaction: &lmdb::RoTxn<'_>) -> tg::Result { + let Some((key, _)) = db + .last(transaction) + .map_err(|error| tg::error!(!error, "failed to get the last outbox entry"))? + else { + return Ok(0); + }; + let seq = u128::from_be_bytes( + key.try_into() + .map_err(|_| tg::error!("invalid outbox key length"))?, + ); + Ok(seq.to_u64().unwrap()) + } +} + +fn seq_to_key(seq: u64) -> [u8; 16] { + u128::from(seq).to_be_bytes() +} + +fn seq_from_key(key: &[u8]) -> tg::Result<[u8; 16]> { + key.try_into() + .map_err(|_| tg::error!("invalid outbox key length")) +} + +fn encode_value(partition: i64, payload: &[u8]) -> Vec { + let mut value = Vec::with_capacity(8 + payload.len()); + value.extend_from_slice(&partition.to_be_bytes()); + value.extend_from_slice(payload); + value +} + +fn decode_partition(value: &[u8]) -> tg::Result { + let bytes = value + .get(..8) + .ok_or_else(|| tg::error!("invalid outbox value length"))?; + Ok(i64::from_be_bytes(bytes.try_into().unwrap())) +} diff --git a/packages/stores/object/src/lmdb/outbox/object.rs b/packages/stores/object/src/lmdb/outbox/object.rs new file mode 100644 index 0000000000..d340443762 --- /dev/null +++ b/packages/stores/object/src/lmdb/outbox/object.rs @@ -0,0 +1,183 @@ +use {super::Item, crate::lmdb::Store, tangram_client::prelude::*}; + +impl Store { + pub async fn enqueue_object_outbox(&self, entries: Vec<(i64, Vec)>) -> tg::Result<()> { + self.enqueue_outbox_inner(self.object_outbox_db, entries) + .await + } + + pub async fn dequeue_object_outbox( + &self, + partitions: Vec, + limit: i32, + ) -> tg::Result> { + self.dequeue_outbox_inner(self.object_outbox_db, partitions, limit) + .await + } + + pub async fn delete_object_outbox(&self, entries: &[(i64, [u8; 16])]) -> tg::Result<()> { + self.delete_outbox_inner(self.object_outbox_db, entries) + .await + } + + pub async fn object_outbox_max_seq( + &self, + partitions: Vec, + ) -> tg::Result> { + self.outbox_max_seq_inner(self.object_outbox_db, partitions) + .await + } + + pub async fn object_outbox_has_pending( + &self, + partitions: Vec, + seq: [u8; 16], + ) -> tg::Result { + self.outbox_has_pending_inner(self.object_outbox_db, partitions, seq) + .await + } +} + +#[cfg(test)] +mod tests { + use crate::lmdb::{Config, Store}; + + fn store(path: &std::path::Path) -> Store { + let config = Config { + map_size: 1024 * 1024 * 10, + path: path.join("test.lmdb"), + }; + Store::new(&config).unwrap() + } + + // Dequeuing returns the enqueued entries in order, filtered by partition, and deleting drains the outbox. + #[tokio::test] + async fn enqueue_dequeue_delete() { + let temp = tangram_util::fs::Temp::new().unwrap(); + std::fs::create_dir(temp.path()).unwrap(); + let store = store(temp.path()); + + store + .enqueue_object_outbox(vec![ + (0, b"a".to_vec()), + (1, b"b".to_vec()), + (0, b"c".to_vec()), + ]) + .await + .unwrap(); + + // Only entries in the requested partitions are dequeued, in enqueue order. + let items = store.dequeue_object_outbox(vec![0], 10).await.unwrap(); + assert_eq!(items.len(), 2); + assert_eq!(items[0].payload.as_ref(), b"a"); + assert_eq!(items[1].payload.as_ref(), b"c"); + + // The limit bounds the number of dequeued entries. + let items = store.dequeue_object_outbox(vec![0, 1], 1).await.unwrap(); + assert_eq!(items.len(), 1); + assert_eq!(items[0].payload.as_ref(), b"a"); + + // Deleting the dequeued entries removes them. + let items = store.dequeue_object_outbox(vec![0, 1], 10).await.unwrap(); + let seqs = items + .iter() + .map(|item| (item.partition, item.seq)) + .collect::>(); + store.delete_object_outbox(&seqs).await.unwrap(); + assert!( + store + .dequeue_object_outbox(vec![0, 1], 10) + .await + .unwrap() + .is_empty() + ); + } + + // The max seq and pending checks reflect the entries in the requested partitions. + #[tokio::test] + async fn max_seq_and_has_pending() { + let temp = tangram_util::fs::Temp::new().unwrap(); + std::fs::create_dir(temp.path()).unwrap(); + let store = store(temp.path()); + + // An empty outbox has no max seq and nothing pending. + assert!( + store + .object_outbox_max_seq(vec![0, 1]) + .await + .unwrap() + .is_none() + ); + + store + .enqueue_object_outbox(vec![(0, b"a".to_vec()), (1, b"b".to_vec())]) + .await + .unwrap(); + let snapshot = store + .object_outbox_max_seq(vec![0, 1]) + .await + .unwrap() + .unwrap(); + + // The snapshot is pending until every entry up to it is drained. + assert!( + store + .object_outbox_has_pending(vec![0, 1], snapshot) + .await + .unwrap() + ); + + // A partition with no entries at or below the snapshot has nothing pending. + let items = store.dequeue_object_outbox(vec![0], 10).await.unwrap(); + let seqs = items + .iter() + .map(|item| (item.partition, item.seq)) + .collect::>(); + store.delete_object_outbox(&seqs).await.unwrap(); + assert!( + !store + .object_outbox_has_pending(vec![0], snapshot) + .await + .unwrap() + ); + + // The remaining partition still has the snapshot pending. + assert!( + store + .object_outbox_has_pending(vec![1], snapshot) + .await + .unwrap() + ); + } + + // Entries persist across a restart and the sequence counter resumes so new sequences remain monotonic. + #[tokio::test] + async fn persists_across_restart() { + let temp = tangram_util::fs::Temp::new().unwrap(); + std::fs::create_dir(temp.path()).unwrap(); + + let snapshot = { + let store = store(temp.path()); + store + .enqueue_object_outbox(vec![(0, b"a".to_vec())]) + .await + .unwrap(); + store.object_outbox_max_seq(vec![0]).await.unwrap().unwrap() + }; + + // Reopen the store and confirm the persisted entry is still present. + let store = store(temp.path()); + assert_eq!( + store.object_outbox_max_seq(vec![0]).await.unwrap(), + Some(snapshot) + ); + + // A newly enqueued entry receives a greater sequence than any persisted before the restart. + store + .enqueue_object_outbox(vec![(0, b"b".to_vec())]) + .await + .unwrap(); + let next = store.object_outbox_max_seq(vec![0]).await.unwrap().unwrap(); + assert!(next > snapshot); + } +} diff --git a/packages/stores/object/src/memory.rs b/packages/stores/object/src/memory.rs index 69f0d1d082..bdeac9ef26 100644 --- a/packages/stores/object/src/memory.rs +++ b/packages/stores/object/src/memory.rs @@ -1,5 +1,6 @@ use { crate::{DeleteArg, Object, PutArg, TryGetArg, TryGetBatchArg, TryGetOutput}, + outbox::Outbox, std::{ collections::HashMap, sync::{Arc, Mutex, MutexGuard}, @@ -10,8 +11,11 @@ use { mod delete; mod flush; mod get; +mod outbox; mod put; +pub use outbox::Item; + #[derive(Clone, Debug, Default)] pub struct Config {} @@ -22,6 +26,7 @@ pub struct Store { #[derive(Default)] struct State { objects: Objects, + object_outbox: Outbox, } type Objects = HashMap, tg::id::BuildHasher>; diff --git a/packages/stores/object/src/memory/outbox.rs b/packages/stores/object/src/memory/outbox.rs new file mode 100644 index 0000000000..a90c876862 --- /dev/null +++ b/packages/stores/object/src/memory/outbox.rs @@ -0,0 +1,68 @@ +use {bytes::Bytes, num::ToPrimitive as _, std::collections::BTreeMap}; + +pub mod object; + +pub struct Item { + pub seq: [u8; 16], + pub partition: i64, + pub payload: Bytes, +} + +#[derive(Default)] +pub(super) struct Outbox { + seq: u128, + entries: BTreeMap<[u8; 16], Entry>, +} + +struct Entry { + partition: i64, + payload: Bytes, +} + +impl Outbox { + pub fn enqueue(&mut self, entries: Vec<(i64, Vec)>) { + for (partition, payload) in entries { + self.seq += 1; + let seq = self.seq.to_be_bytes(); + let entry = Entry { + partition, + payload: payload.into(), + }; + self.entries.insert(seq, entry); + } + } + + pub fn dequeue(&self, partitions: &[i64], limit: i32) -> Vec { + let limit = limit.to_usize().unwrap_or(0); + self.entries + .iter() + .filter(|(_, entry)| partitions.contains(&entry.partition)) + .take(limit) + .map(|(seq, entry)| Item { + seq: *seq, + partition: entry.partition, + payload: entry.payload.clone(), + }) + .collect() + } + + pub fn delete(&mut self, entries: &[(i64, [u8; 16])]) { + for (_, seq) in entries { + self.entries.remove(seq); + } + } + + pub fn max_seq(&self, partitions: &[i64]) -> Option<[u8; 16]> { + self.entries + .iter() + .rev() + .find(|(_, entry)| partitions.contains(&entry.partition)) + .map(|(seq, _)| *seq) + } + + pub fn has_pending(&self, partitions: &[i64], seq: [u8; 16]) -> bool { + self.entries + .range(..=seq) + .any(|(_, entry)| partitions.contains(&entry.partition)) + } +} diff --git a/packages/stores/object/src/memory/outbox/object.rs b/packages/stores/object/src/memory/outbox/object.rs new file mode 100644 index 0000000000..0b91aeb592 --- /dev/null +++ b/packages/stores/object/src/memory/outbox/object.rs @@ -0,0 +1,90 @@ +use {super::Item, crate::memory::Store}; + +impl Store { + pub fn enqueue_object_outbox(&self, entries: Vec<(i64, Vec)>) { + self.state().object_outbox.enqueue(entries); + } + + #[must_use] + pub fn dequeue_object_outbox(&self, partitions: &[i64], limit: i32) -> Vec { + self.state().object_outbox.dequeue(partitions, limit) + } + + pub fn delete_object_outbox(&self, entries: &[(i64, [u8; 16])]) { + self.state().object_outbox.delete(entries); + } + + #[must_use] + pub fn object_outbox_max_seq(&self, partitions: &[i64]) -> Option<[u8; 16]> { + self.state().object_outbox.max_seq(partitions) + } + + #[must_use] + pub fn object_outbox_has_pending(&self, partitions: &[i64], seq: [u8; 16]) -> bool { + self.state().object_outbox.has_pending(partitions, seq) + } +} + +#[cfg(test)] +mod tests { + use crate::memory::Store; + + // Dequeuing returns the enqueued entries in order, filtered by partition, and deleting drains the outbox. + #[test] + fn enqueue_dequeue_delete() { + let store = Store::new(); + + store.enqueue_object_outbox(vec![ + (0, b"a".to_vec()), + (1, b"b".to_vec()), + (0, b"c".to_vec()), + ]); + + // Only entries in the requested partitions are dequeued, in enqueue order. + let items = store.dequeue_object_outbox(&[0], 10); + assert_eq!(items.len(), 2); + assert_eq!(items[0].payload.as_ref(), b"a"); + assert_eq!(items[1].payload.as_ref(), b"c"); + + // The limit bounds the number of dequeued entries. + let items = store.dequeue_object_outbox(&[0, 1], 1); + assert_eq!(items.len(), 1); + assert_eq!(items[0].payload.as_ref(), b"a"); + + // Deleting the dequeued entries removes them. + let items = store.dequeue_object_outbox(&[0, 1], 10); + let seqs = items + .iter() + .map(|item| (item.partition, item.seq)) + .collect::>(); + store.delete_object_outbox(&seqs); + assert!(store.dequeue_object_outbox(&[0, 1], 10).is_empty()); + } + + // The max seq and pending checks reflect the entries in the requested partitions. + #[test] + fn max_seq_and_has_pending() { + let store = Store::new(); + + // An empty outbox has no max seq and nothing pending. + assert!(store.object_outbox_max_seq(&[0, 1]).is_none()); + + store.enqueue_object_outbox(vec![(0, b"a".to_vec()), (1, b"b".to_vec())]); + let snapshot = store.object_outbox_max_seq(&[0, 1]).unwrap(); + + // The snapshot is pending until every entry up to it is drained. + assert!(store.object_outbox_has_pending(&[0, 1], snapshot)); + + // A partition with no entries at or below the snapshot has nothing pending. + let items = store.dequeue_object_outbox(&[0], 10); + let seqs = items + .iter() + .map(|item| (item.partition, item.seq)) + .collect::>(); + store.delete_object_outbox(&seqs); + assert!(!store.object_outbox_has_pending(&[0], snapshot)); + + // The remaining partition still has the snapshot pending. + assert!(store.object_outbox_has_pending(&[1], snapshot)); + } +} diff --git a/packages/stores/object/src/scylla.cql b/packages/stores/object/src/scylla.cql index 7fbfd53c6a..4611cf4f00 100644 --- a/packages/stores/object/src/scylla.cql +++ b/packages/stores/object/src/scylla.cql @@ -4,3 +4,10 @@ create table if not exists objects ( id blob primary key, stored_at bigint ); + +create table if not exists object_outbox ( + partition bigint, + seq timeuuid, + payload blob, + primary key ((partition), seq) +); diff --git a/packages/stores/object/src/scylla.rs b/packages/stores/object/src/scylla.rs index 6d23068757..457d97be5d 100644 --- a/packages/stores/object/src/scylla.rs +++ b/packages/stores/object/src/scylla.rs @@ -8,8 +8,11 @@ use { mod delete; mod flush; mod get; +mod outbox; mod put; +pub use outbox::Item; + #[derive(Clone, Debug)] pub struct Config { pub addr: String, @@ -39,8 +42,13 @@ pub struct Store { struct Statements { delete_object: scylla::statement::prepared::PreparedStatement, + delete_object_outbox: scylla::statement::prepared::PreparedStatement, + dequeue_object_outbox: scylla::statement::prepared::PreparedStatement, + enqueue_object_outbox: scylla::statement::prepared::PreparedStatement, get_object_batch: scylla::statement::prepared::PreparedStatement, get_object: scylla::statement::prepared::PreparedStatement, + object_outbox_max_seq: scylla::statement::prepared::PreparedStatement, + object_outbox_pending: scylla::statement::prepared::PreparedStatement, put_object: scylla::statement::prepared::PreparedStatement, } @@ -140,11 +148,80 @@ impl Store { .map_err(|error| tg::error!(!error, "failed to prepare the put statement"))?; put_object.set_consistency(scylla::statement::Consistency::LocalQuorum); + let statement = indoc!( + " + insert into object_outbox (partition, seq, payload) + values (?, now(), ?); + " + ); + let mut enqueue_object_outbox = session.prepare(statement).await.map_err(|error| { + tg::error!(!error, "failed to prepare the enqueue outbox statement") + })?; + enqueue_object_outbox.set_consistency(scylla::statement::Consistency::LocalQuorum); + + let statement = indoc!( + " + select partition, seq, payload + from object_outbox + where partition in ? + limit ?; + " + ); + let mut dequeue_object_outbox = session.prepare(statement).await.map_err(|error| { + tg::error!( + !error, + "failed to prepare the dequeue outbox batch statement" + ) + })?; + dequeue_object_outbox.set_consistency(scylla::statement::Consistency::One); + + let statement = indoc!( + " + delete from object_outbox + where partition = ? and seq = ?; + " + ); + let mut delete_object_outbox = session + .prepare(statement) + .await + .map_err(|error| tg::error!(!error, "failed to prepare the delete outbox statement"))?; + delete_object_outbox.set_consistency(scylla::statement::Consistency::LocalQuorum); + + let statement = indoc!( + " + select max(seq) + from object_outbox + where partition in ?; + " + ); + let mut object_outbox_max_seq = session.prepare(statement).await.map_err(|error| { + tg::error!(!error, "failed to prepare the outbox max seq statement") + })?; + object_outbox_max_seq.set_consistency(scylla::statement::Consistency::LocalQuorum); + + let statement = indoc!( + " + select seq + from object_outbox + where partition in ? and seq <= ? + limit 1; + " + ); + let mut object_outbox_pending = session.prepare(statement).await.map_err(|error| { + tg::error!(!error, "failed to prepare the outbox pending statement") + })?; + object_outbox_pending.set_consistency(scylla::statement::Consistency::LocalQuorum); + let scylla = Self { statements: Statements { delete_object, + delete_object_outbox, + dequeue_object_outbox, + enqueue_object_outbox, get_object_batch, get_object, + object_outbox_max_seq, + object_outbox_pending, put_object, }, session, diff --git a/packages/stores/object/src/scylla/outbox.rs b/packages/stores/object/src/scylla/outbox.rs new file mode 100644 index 0000000000..e4323b6bad --- /dev/null +++ b/packages/stores/object/src/scylla/outbox.rs @@ -0,0 +1,51 @@ +use { + super::Store, bytes::Bytes, futures::FutureExt as _, scylla::value::CqlTimeuuid, + tangram_client::prelude::*, +}; + +pub mod object; + +pub struct Item { + pub seq: [u8; 16], + pub partition: i64, + pub payload: Bytes, +} + +impl Store { + pub(super) async fn dequeue_outbox_with_statement( + &self, + partitions: &[i64], + limit: i32, + statement: &scylla::statement::prepared::PreparedStatement, + ) -> tg::Result> { + let params = (partitions, limit); + #[derive(scylla::DeserializeRow)] + struct Row<'a> { + partition: i64, + seq: CqlTimeuuid, + payload: &'a [u8], + } + let result = self + .session + .execute_unpaged(statement, params) + .boxed() + .await + .map_err(|error| tg::error!(!error, "failed to execute the query"))? + .into_rows_result() + .map_err(|error| tg::error!(!error, "failed to get the rows"))?; + let items = result + .rows::() + .map_err(|error| tg::error!(!error, "failed to iterate the rows"))? + .map(|result| { + let row = result.map_err(|error| tg::error!(!error, "failed to get the row"))?; + let payload = Bytes::copy_from_slice(row.payload); + Ok(Item { + seq: *row.seq.as_bytes(), + partition: row.partition, + payload, + }) + }) + .collect::>>()?; + Ok(items) + } +} diff --git a/packages/stores/object/src/scylla/outbox/object.rs b/packages/stores/object/src/scylla/outbox/object.rs new file mode 100644 index 0000000000..255409ef96 --- /dev/null +++ b/packages/stores/object/src/scylla/outbox/object.rs @@ -0,0 +1,114 @@ +use {super::Item, crate::scylla::Store, scylla::value::CqlTimeuuid, tangram_client::prelude::*}; + +impl Store { + pub async fn enqueue_object_outbox(&self, entries: Vec<(i64, Vec)>) -> tg::Result<()> { + if entries.is_empty() { + return Ok(()); + } + let mut batch = + scylla::statement::batch::Batch::new(scylla::statement::batch::BatchType::Unlogged); + batch.set_consistency( + self.statements + .enqueue_object_outbox + .get_consistency() + .unwrap(), + ); + for _ in &entries { + batch.append_statement(scylla::statement::batch::BatchStatement::PreparedStatement( + self.statements.enqueue_object_outbox.clone(), + )); + } + self.session + .batch(&batch, &entries) + .await + .map_err(|error| tg::error!(!error, "failed to execute the batch"))?; + Ok(()) + } + + pub async fn dequeue_object_outbox( + &self, + partitions: Vec, + limit: i32, + ) -> tg::Result> { + let items = self + .dequeue_outbox_with_statement( + &partitions, + limit, + &self.statements.dequeue_object_outbox, + ) + .await?; + if !items.is_empty() { + return Ok(items); + } + let mut statement = self.statements.dequeue_object_outbox.clone(); + statement.set_consistency(scylla::statement::Consistency::LocalQuorum); + self.dequeue_outbox_with_statement(&partitions, limit, &statement) + .await + } + + pub async fn delete_object_outbox(&self, entries: &[(i64, [u8; 16])]) -> tg::Result<()> { + if entries.is_empty() { + return Ok(()); + } + let mut batch = + scylla::statement::batch::Batch::new(scylla::statement::batch::BatchType::Unlogged); + batch.set_consistency( + self.statements + .delete_object_outbox + .get_consistency() + .unwrap(), + ); + for _ in entries { + batch.append_statement(scylla::statement::batch::BatchStatement::PreparedStatement( + self.statements.delete_object_outbox.clone(), + )); + } + let params = entries + .iter() + .map(|(partition, seq)| (*partition, CqlTimeuuid::from_bytes(*seq))) + .collect::>(); + self.session + .batch(&batch, params) + .await + .map_err(|error| tg::error!(!error, "failed to delete the outbox entries"))?; + Ok(()) + } + + pub async fn object_outbox_max_seq( + &self, + partitions: Vec, + ) -> tg::Result> { + let params = (partitions,); + let result = self + .session + .execute_unpaged(&self.statements.object_outbox_max_seq, params) + .await + .map_err(|error| tg::error!(!error, "failed to execute the query"))? + .into_rows_result() + .map_err(|error| tg::error!(!error, "failed to get the rows"))?; + let seq = result + .maybe_first_row::<(Option,)>() + .map_err(|error| tg::error!(!error, "failed to get the row"))? + .and_then(|(seq,)| seq); + Ok(seq.map(|seq| *seq.as_bytes())) + } + + pub async fn object_outbox_has_pending( + &self, + partitions: Vec, + seq: [u8; 16], + ) -> tg::Result { + let params = (partitions, CqlTimeuuid::from_bytes(seq)); + let result = self + .session + .execute_unpaged(&self.statements.object_outbox_pending, params) + .await + .map_err(|error| tg::error!(!error, "failed to execute the query"))? + .into_rows_result() + .map_err(|error| tg::error!(!error, "failed to get the rows"))?; + let row = result + .maybe_first_row::<(CqlTimeuuid,)>() + .map_err(|error| tg::error!(!error, "failed to get the row"))?; + Ok(row.is_some()) + } +} diff --git a/scripts/cloud/kubernetes.yaml b/scripts/cloud/kubernetes.yaml index 146e2e7d0f..f5433e6632 100644 --- a/scripts/cloud/kubernetes.yaml +++ b/scripts/cloud/kubernetes.yaml @@ -125,6 +125,8 @@ spec: args: - --overprovisioned - "1" + - --max-partition-key-restrictions-per-query + - "1024" ports: - containerPort: 9042 readinessProbe: