feat: object outbox

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

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

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

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"

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<crate::grant::delete::Arg>,
+	#[serde(default, skip_serializing_if = "Vec::is_empty")]
 	pub delete_group_members: Vec<crate::group::member::delete::Arg>,
+	#[serde(default, skip_serializing_if = "Vec::is_empty")]
 	pub delete_groups: Vec<tg::group::Id>,
+	#[serde(default, skip_serializing_if = "Vec::is_empty")]
 	pub delete_organization_members: Vec<crate::organization::member::delete::Arg>,
+	#[serde(default, skip_serializing_if = "Vec::is_empty")]
 	pub delete_organizations: Vec<tg::organization::Id>,
+	#[serde(default, skip_serializing_if = "Vec::is_empty")]
 	pub delete_tags: Vec<tg::tag::Id>,
+	#[serde(default, skip_serializing_if = "Vec::is_empty")]
 	pub delete_users: Vec<tg::user::Id>,
+	#[serde(default, skip_serializing_if = "Vec::is_empty")]
 	pub put_cache_entries: Vec<crate::cache::put::Arg>,
+	#[serde(default, skip_serializing_if = "Vec::is_empty")]
 	pub put_grants: Vec<crate::grant::put::Arg>,
+	#[serde(default, skip_serializing_if = "Vec::is_empty")]
 	pub put_group_members: Vec<crate::group::member::put::Arg>,
+	#[serde(default, skip_serializing_if = "Vec::is_empty")]
 	pub put_groups: Vec<crate::group::put::Arg>,
+	#[serde(default, skip_serializing_if = "Vec::is_empty")]
 	pub put_objects: Vec<crate::object::put::Arg>,
+	#[serde(default, skip_serializing_if = "Vec::is_empty")]
 	pub put_organization_members: Vec<crate::organization::member::put::Arg>,
+	#[serde(default, skip_serializing_if = "Vec::is_empty")]
 	pub put_organizations: Vec<crate::organization::put::Arg>,
+	#[serde(default, skip_serializing_if = "Vec::is_empty")]
 	pub put_processes: Vec<crate::process::put::Arg>,
+	#[serde(default, skip_serializing_if = "Vec::is_empty")]
 	pub put_sandboxes: Vec<crate::sandbox::put::Arg>,
+	#[serde(default, skip_serializing_if = "Vec::is_empty")]
 	pub put_tags: Vec<crate::tag::put::Arg>,
+	#[serde(default, skip_serializing_if = "Vec::is_empty")]
 	pub put_users: Vec<crate::user::put::Arg>,
 }
 

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<tg::artifact::Id>,
 	pub id: tg::artifact::Id,

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<tg::Principal>,
 	pub expires_at: Option<i64>,

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<tg::Principal>,

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,

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,

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<tg::Id>,

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<tg::artifact::Id>,
 	pub children: BTreeSet<tg::object::Id>,

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,

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,

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,

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<Vec<tg::process::Id>>,
 	pub command: tg::object::Id,

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<tg::Principal>,

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<tg::object::Id, tg::process::Id>,

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,

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

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