feat(server/runner): sandbox/process control connection pools

packages/cli/src/config.rs

@@ -1352,6 +1352,10 @@ pub struct RemoteCache {
 #[derive(Clone, Debug, Default, serde::Deserialize, serde::Serialize)]
 #[serde(deny_unknown_fields)]
 pub struct Runner {
+	#[serde_as(as = "Option<DurationSecondsWithFrac>")]
+	#[serde(default, skip_serializing_if = "Option::is_none")]
+	pub connection_pool_refill_interval: Option<Duration>,
+
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub cpus: Option<u64>,
 
@@ -1368,6 +1372,9 @@ pub struct Runner {
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub memory: Option<u64>,
 
+	#[serde(default, skip_serializing_if = "Option::is_none")]
+	pub process_control_pool_size: Option<usize>,
+
 	#[serde_as(as = "Option<DurationSecondsWithFrac>")]
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub process_state_ttl: Option<Duration>,
@@ -1379,6 +1386,9 @@ pub struct Runner {
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub remote: Option<String>,
 
+	#[serde(default, skip_serializing_if = "Option::is_none")]
+	pub sandbox_control_pool_size: Option<usize>,
+
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub sandbox_pool_size: Option<usize>,
 
@@ -3664,15 +3674,24 @@ fn resolve_runner(source: Runner) -> server::Runner {
 	if let Some(source) = source.js {
 		target.js = resolve_js(source);
 	}
+	if let Some(value) = source.connection_pool_refill_interval {
+		target.connection_pool_refill_interval = value;
+	}
 	if let Some(value) = source.heartbeat_interval {
 		target.heartbeat_interval = value;
 	}
+	if let Some(value) = source.process_control_pool_size {
+		target.process_control_pool_size = value;
+	}
 	if let Some(value) = source.process_state_ttl {
 		target.process_state_ttl = value;
 	}
 	if let Some(value) = source.progress_log_delay {
 		target.progress_log_delay = value;
 	}
+	if let Some(value) = source.sandbox_control_pool_size {
+		target.sandbox_control_pool_size = value;
+	}
 	if let Some(value) = source.sandbox_pool_size {
 		target.sandbox_pool_size = value;
 	}

packages/cli/tests/runner/control_pool_refills.nu

@@ -0,0 +1,51 @@
+use ../../test.nu *
+
+# The control pools reserve a replacement connection on the refill interval after one is taken.
+
+let root_token = random chars
+let remote = server spawn --preserve-keys --name remote --config {
+	advanced: { checkpoints: true, single_process: false },
+	authentication: { root: { token: $root_token }, users: { providers: { insecure: true } } },
+	roles: [api indexer scheduler],
+}
+let created = tg --url $remote.url --token $root_token runner create | from json
+let runner = server spawn --name runner --config {
+	advanced: { checkpoints: true },
+	remotes: { default: { token: $created.token.token, url: $remote.url } },
+	roles: [api indexer runner],
+	runner: {
+		connection_pool_refill_interval: 0.1,
+		id: $created.data.id,
+		process_control_pool_size: 1,
+		remote: 'default',
+		sandbox_control_pool_size: 1,
+		token: $created.token.token,
+	},
+}
+let alice = tg --url $remote.url login --verbose --name alice | from json
+let local = server spawn --name alice-local --config {
+	remotes: { default: { token: $alice.token, url: $remote.url } },
+}
+
+let sandbox_reserved_watch = tg --url $runner.url checkpoint watch runner.sandbox.control.reserved | from json | get watch
+let process_reserved_watch = tg --url $runner.url checkpoint watch runner.process.control.reserved | from json | get watch
+
+# A shortcut child takes both reserved connections.
+let path = artifact {
+	"example.tg.ts": '
+		export default () => tg.run(child).sandbox(true);
+		export const child = () => 42;
+	'
+}
+let output = tg --url $local.url run --no-tty --remote --user $alice.user.id $"($path)/example.tg.ts" | complete
+success $output "the run should succeed with pooled control connections"
+
+# Each pool reserves a replacement on the refill interval.
+let output = timeout 30s tg --url $runner.url checkpoint wait runner.sandbox.control.reserved $sandbox_reserved_watch 0 | complete
+success $output "the sandbox control pool should refill after a take"
+tg --url $runner.url checkpoint continue runner.sandbox.control.reserved $sandbox_reserved_watch 0
+tg --url $runner.url checkpoint unwatch runner.sandbox.control.reserved $sandbox_reserved_watch
+let output = timeout 30s tg --url $runner.url checkpoint wait runner.process.control.reserved $process_reserved_watch 0 | complete
+success $output "the process control pool should refill after a take"
+tg --url $runner.url checkpoint continue runner.process.control.reserved $process_reserved_watch 0
+tg --url $runner.url checkpoint unwatch runner.process.control.reserved $process_reserved_watch

packages/cli/tests/runner/reserved_control_requires_runner.nu

@@ -0,0 +1,19 @@
+use ../../test.nu *
+
+# A reserved process or sandbox control connection is refused for a principal that is not a runner.
+
+let root_token = random chars
+let remote = server spawn --name remote --config {
+	authentication: { root: { token: $root_token } },
+	roles: [api indexer scheduler],
+}
+let socket = $remote.url | str replace 'http+unix://' '' | url decode
+let headers = { Authorization: $'Bearer ($root_token)', 'Content-Type': 'application/vnd.tangram.process-control' }
+let output = http post --full --allow-errors --max-time 10sec --unix-socket $socket --headers $headers 'http://localhost/processes/control?reserved=true' ''
+assert equal $output.status 500 "a root principal must not reserve a process control connection"
+assert ($output.body | to text | str contains 'requires a runner') ($output.body | to text)
+
+let headers = { Authorization: $'Bearer ($root_token)', 'Content-Type': 'text/event-stream' }
+let output = http post --full --allow-errors --max-time 10sec --unix-socket $socket --headers $headers 'http://localhost/sandboxes/control?reserved=true' ''
+assert equal $output.status 500 "a root principal must not reserve a sandbox control connection"
+assert ($output.body | to text | str contains 'requires a runner') ($output.body | to text)

packages/cli/tests/runner/reserved_control_start_replays_after_reconnect.nu

@@ -0,0 +1,54 @@
+use ../../test.nu *
+
+# A reserved connection whose start the remote has not processed reconnects after a remote restart and replays the start.
+
+let root_token = random chars
+let remote = server spawn --preserve-keys --name remote --config {
+	advanced: { checkpoints: true, single_process: false },
+	authentication: { root: { token: $root_token }, users: { providers: { insecure: true } } },
+	roles: [api indexer scheduler],
+}
+let created = tg --url $remote.url --token $root_token runner create | from json
+let runner = server spawn --name runner --config {
+	advanced: { checkpoints: true },
+	remotes: { default: { token: $created.token.token, url: $remote.url } },
+	roles: [api indexer runner],
+	runner: {
+		id: $created.data.id,
+		process_control_pool_size: 1,
+		remote: 'default',
+		sandbox_control_pool_size: 1,
+		token: $created.token.token,
+	},
+}
+let alice = tg --url $remote.url login --verbose --name alice | from json
+let local = server spawn --name alice-local --config {
+	remotes: { default: { token: $alice.token, url: $remote.url } },
+}
+
+# Hold both starts on the remote.
+let sandbox_start_watch = tg --url $remote.url --token $root_token checkpoint watch sandbox.control.start.started | from json | get watch
+let process_start_watch = tg --url $remote.url --token $root_token checkpoint watch process.control.start.started | from json | get watch
+
+let path = artifact {
+	"example.tg.ts": '
+		export default () => tg.run(child).sandbox(true);
+		export const child = () => 42;
+	'
+}
+let spawned = tg --url $local.url build --remote --detach --verbose --user $alice.user.id $"($path)/example.tg.ts" | from json
+let process = $spawned.process | split row '?' | first
+
+# The remote receives both starts and is held before indexing either.
+let output = timeout 30s tg --url $remote.url --token $root_token checkpoint wait sandbox.control.start.started $sandbox_start_watch 0 | complete
+success $output "the remote should receive the sandbox start"
+let output = timeout 30s tg --url $remote.url --token $root_token checkpoint wait process.control.start.started $process_start_watch 0 | complete
+success $output "the remote should receive the process start"
+
+# Restart the remote. The reserved connections reconnect and replay their unacknowledged starts.
+let remote = server restart $remote
+
+let output = timeout 60s tg --url $local.url wait --remote $process | complete
+success $output "the run should succeed after the starts replay"
+let output = $output.stdout | from json
+assert equal $output.exit 0 "the run should finish successfully after the starts replay"

packages/cli/tests/runner/shortcut_child_finishes_before_start.nu

@@ -0,0 +1,88 @@
+use ../../test.nu *
+
+# A shortcut child that finishes before the runner has started it on the remote must still deliver its result to the guest client.
+
+if $nu.os-info.name != 'linux' {
+	skip_test 'this test requires linux'
+}
+
+let root_token = random chars
+
+# Spawn the remote and create the runner.
+let remote = server spawn --preserve-keys --name remote --config {
+	advanced: { checkpoints: true, single_process: false },
+	authentication: { root: { token: $root_token }, users: { providers: { insecure: true } } },
+	roles: [api indexer scheduler],
+}
+let created = tg --url $remote.url --token $root_token runner create | from json
+
+# Spawn the runner with the control pools, as deployed.
+let runner = server spawn --name runner --config {
+	advanced: { checkpoints: true },
+	remotes: { default: { token: $created.token.token, url: $remote.url } },
+	roles: [api indexer runner],
+	runner: {
+		id: $created.data.id,
+		process_control_pool_size: 8,
+		remote: 'default',
+		sandbox_control_pool_size: 8,
+		sandbox_pool_size: 8,
+		token: $created.token.token,
+	},
+}
+
+let dir = tg --url $remote.url --token $root_token put -k directory 'tg.directory({})' | str trim
+
+# Hold the child's command push so the remote does not learn about the child, and hold the spawn reply so the child finishes before the guest client can wait for it.
+let push_watch = (
+	tg --url $runner.url checkpoint watch runner.process.command.push.started
+	| from json
+	| get watch
+)
+let spawn_watch = (
+	tg --url $runner.url checkpoint watch process.spawn.child.add
+	| from json
+	| get watch
+)
+let finished_watch = (
+	tg --url $runner.url checkpoint watch runner.process.finished
+	| from json
+	| get watch
+)
+
+# Run a sandboxed parent that spawns a trivial child through the guest URL.
+let run = job spawn {
+	let job_id = job id
+	let output = tg --url $remote.url --token $root_token run --sandbox --executable /bin/sh $dir -- -c '/opt/tangram/bin/tangram run --sandbox --executable /bin/sh "$0" -- -c "echo hello-from-child"' $dir | complete
+	$output | job send --tag $job_id 0
+}
+
+# The child's push and spawn reply are held while the child runs to completion.
+let output = timeout 60s tg --url $runner.url checkpoint wait runner.process.command.push.started $push_watch 0 | complete
+success $output "the child should reach its command push"
+let output = timeout 60s tg --url $runner.url checkpoint wait process.spawn.child.add $spawn_watch 0 | complete
+success $output "the child spawn should reach its reply"
+let output = timeout 60s tg --url $runner.url checkpoint wait runner.process.finished $finished_watch 0 | complete
+success $output "the child should finish while its push and spawn reply are held"
+tg --url $runner.url checkpoint continue runner.process.finished $finished_watch 0
+
+# Release the spawn reply: the guest client now waits for a child that finished before the remote knows it.
+tg --url $runner.url checkpoint continue process.spawn.child.add $spawn_watch 0
+tg --url $runner.url checkpoint unwatch process.spawn.child.add $spawn_watch
+
+# The parent finishes only after the child's result reached the guest client.
+let output = timeout 60s tg --url $runner.url checkpoint wait runner.process.finished $finished_watch 1 | complete
+success $output "the parent should finish"
+tg --url $runner.url checkpoint continue runner.process.finished $finished_watch 1
+tg --url $runner.url checkpoint unwatch runner.process.finished $finished_watch
+
+# Release the push.
+tg --url $runner.url checkpoint continue runner.process.command.push.started $push_watch 0
+tg --url $runner.url checkpoint unwatch runner.process.command.push.started $push_watch
+
+let output = try { job recv --tag $run --timeout 60sec } catch { null }
+if $output == null {
+	error make { msg: "the run did not complete" }
+}
+success $output "the child spawned from inside the sandbox should succeed when it finishes before the remote starts it"
+assert ($output.stdout | str contains "hello-from-child") "the child output should reach the client"

packages/cli/tests/runner/shortcut_control_failure.nu

@@ -14,7 +14,7 @@ for kind in [sandbox process] {
 		advanced: { checkpoints: true },
 		remotes: { default: { token: $created.token.token, url: $remote.url } },
 		roles: [api indexer runner],
-		runner: { cpus: 1, id: $created.data.id, remote: default, sandbox_pool_size: 1, token: $created.token.token },
+		runner: { cpus: 1, id: $created.data.id, process_control_pool_size: 0, remote: default, sandbox_control_pool_size: 0, sandbox_pool_size: 1, token: $created.token.token },
 	}
 	let alice = tg --url $remote.url login --verbose --name alice | from json
 	let local = server spawn --name $'local-($kind)' --config {

packages/cli/tests/runner/shortcut_process_uses_pooled_control.nu

@@ -0,0 +1,110 @@
+use ../../test.nu *
+
+# A shortcut child in a new sandbox takes reserved sandbox and process control connections from the pools. It runs and finishes before the control server starts either of them.
+
+let root_token = random chars
+
+# Spawn the remote and create the runner.
+let remote = server spawn --preserve-keys --name remote --config {
+	advanced: { checkpoints: true, single_process: false },
+	authentication: { root: { token: $root_token }, users: { providers: { insecure: true } } },
+	roles: [api indexer scheduler],
+}
+let created = tg --url $remote.url --token $root_token runner create | from json
+
+# Spawn the runner with sandbox and process control pools.
+let runner = server spawn --name runner --config {
+	advanced: { checkpoints: true },
+	remotes: { default: { token: $created.token.token, url: $remote.url } },
+	roles: [api indexer runner],
+	runner: {
+		id: $created.data.id,
+		process_control_pool_size: 1,
+		remote: 'default',
+		sandbox_control_pool_size: 1,
+		token: $created.token.token,
+	},
+}
+
+# Create user credentials and spawn the local server.
+let alice = tg --url $remote.url login --verbose --name alice | from json
+let local = server spawn --name alice-local --config {
+	remotes: { default: { token: $alice.token, url: $remote.url } },
+}
+
+# Hold the starts on the remote so the child runs while the server has not started its sandbox or process.
+let sandbox_start_watch = (
+	tg --url $remote.url --token $root_token checkpoint watch sandbox.control.start.started
+	| from json
+	| get watch
+)
+let process_start_watch = (
+	tg --url $remote.url --token $root_token checkpoint watch process.control.start.started
+	| from json
+	| get watch
+)
+
+# Watch the pooled starts and the child finish on the runner.
+let sandbox_sent_watch = (
+	tg --url $runner.url checkpoint watch runner.sandbox.control.start.sent
+	| from json
+	| get watch
+)
+let process_sent_watch = (
+	tg --url $runner.url checkpoint watch runner.process.control.start.sent
+	| from json
+	| get watch
+)
+let finish_watch = (
+	tg --url $runner.url checkpoint watch runner.process.finish
+	| from json
+	| get watch
+)
+
+let path = artifact {
+	"example.tg.ts": '
+		export default () => tg.run(child).sandbox(true);
+		export const child = () => 42;
+	'
+}
+let build = job spawn {
+	let job_id = job id
+	let output = tg --url $local.url run --no-tty --remote --user $alice.user.id $"($path)/example.tg.ts" | complete
+	$output | job send --tag $job_id 0
+}
+
+# The shortcut child starts the pooled sandbox and process connections.
+let output = timeout 30s tg --url $runner.url checkpoint wait runner.sandbox.control.start.sent $sandbox_sent_watch 0 | complete
+success $output "the shortcut child should start a pooled sandbox control connection"
+tg --url $runner.url checkpoint continue runner.sandbox.control.start.sent $sandbox_sent_watch 0
+tg --url $runner.url checkpoint unwatch runner.sandbox.control.start.sent $sandbox_sent_watch
+
+let output = timeout 30s tg --url $runner.url checkpoint wait runner.process.control.start.sent $process_sent_watch 0 | complete
+success $output "the shortcut child should start a pooled process control connection"
+tg --url $runner.url checkpoint continue runner.process.control.start.sent $process_sent_watch 0
+tg --url $runner.url checkpoint unwatch runner.process.control.start.sent $process_sent_watch
+
+# The remote receives both starts and is held there.
+let output = timeout 30s tg --url $remote.url --token $root_token checkpoint wait sandbox.control.start.started $sandbox_start_watch 0 | complete
+success $output "the remote should receive the sandbox start"
+let output = timeout 30s tg --url $remote.url --token $root_token checkpoint wait process.control.start.started $process_start_watch 0 | complete
+success $output "the remote should receive the process start"
+
+# The child finishes on the runner while the remote has not started its sandbox or process.
+let output = timeout 30s tg --url $runner.url checkpoint wait runner.process.finish $finish_watch 0 | complete
+success $output "the child should finish before the remote starts its sandbox and process"
+tg --url $runner.url checkpoint continue runner.process.finish $finish_watch 0
+tg --url $runner.url checkpoint unwatch runner.process.finish $finish_watch
+
+# Release the remote.
+tg --url $remote.url --token $root_token checkpoint continue sandbox.control.start.started $sandbox_start_watch 0
+tg --url $remote.url --token $root_token checkpoint unwatch sandbox.control.start.started $sandbox_start_watch
+tg --url $remote.url --token $root_token checkpoint continue process.control.start.started $process_start_watch 0
+tg --url $remote.url --token $root_token checkpoint unwatch process.control.start.started $process_start_watch
+
+let output = try { job recv --tag $build --timeout 30sec } catch { null }
+if $output == null {
+	error make { msg: "the run did not complete" }
+}
+success $output "the run should succeed with pooled control connections"
+assert ($output.stdout | str contains "42") "the run should return the child output"