feat(server): add file system monitor

- Populate artifact_paths table during check ins.
- Add file watching via the notify crate to track changes to invalidate the artifact_paths table.
- Added commands for inspecting/manipulating the current watches.
- Added server configuration to disable the file system monitor.

Author: m-hilgendorf

Cargo.lock

@@ -68,6 +68,7 @@ indicatif = "0.17"
 indoc = "2"
 itertools = "0.12"
 libc = "0.2"
+lru = "0.12"
 lsp-types = { version = "0.95" }
 mime = "0.3"
 notify = "6"

Cargo.toml

@@ -38,6 +38,7 @@ serde_json = { workspace = true }
 serde_with = { workspace = true }
 tangram_client = { workspace = true }
 tangram_server = { workspace = true }
+time = { workspace = true }
 tokio = { workspace = true }
 tokio-util = { workspace = true }
 tracing = { workspace = true }

packages/cli/Cargo.toml

@@ -23,3 +23,4 @@ pub mod run;
 pub mod server;
 pub mod tree;
 pub mod upgrade;
+pub mod watch;

packages/cli/src/commands.rs

@@ -7,6 +7,10 @@ use tangram_client as tg;
 pub struct Args {
 	/// The path to check in.
 	pub path: Option<PathBuf>,
+
+	/// Toggle whether the server will add a file watcher for this path.
+	#[arg(long)]
+	pub watch: Option<bool>,
 }
 
 impl Cli {
@@ -19,9 +23,10 @@ impl Cli {
 		if let Some(path_arg) = &args.path {
 			path.push(path_arg);
 		}
+		let watch = args.watch.unwrap_or(true);
 
 		// Perform the checkin.
-		let artifact = tg::Artifact::check_in(client, &path.try_into()?).await?;
+		let artifact = tg::Artifact::check_in(client, &path.try_into()?, watch).await?;
 
 		// Print the ID.
 		let id = artifact.id(client).await?;

packages/cli/src/commands/checkin.rs

@@ -140,6 +140,17 @@ impl Cli {
 				},
 			);
 
+		// Create the file system monitor options
+		let file_system_monitor = config
+			.as_ref()
+			.and_then(|config| config.file_system_monitor.as_ref())
+			.map_or(
+				tangram_server::options::FileSystemMonitor { enable: true },
+				|file_system_monitor| tangram_server::options::FileSystemMonitor {
+					enable: file_system_monitor.enable,
+				},
+			);
+
 		// Create the messenger options.
 		let messenger = config
 			.and_then(|config| config.messenger.as_ref())
@@ -248,6 +259,7 @@ impl Cli {
 			advanced,
 			build,
 			database,
+			file_system_monitor,
 			messenger,
 			oauth,
 			path,

packages/cli/src/commands/server.rs

@@ -0,0 +1,79 @@
+use std::path::PathBuf;
+
+use tangram_client as tg;
+
+use crate::Cli;
+use tg::Handle;
+
+/// Manage watches.
+#[derive(Debug, clap::Args)]
+pub struct Args {
+	#[clap(subcommand)]
+	pub command: Command,
+}
+
+#[derive(Debug, clap::Subcommand)]
+pub enum Command {
+	Add(AddArgs),
+	Remove(RemoveArgs),
+	List(ListArgs),
+}
+
+#[derive(Debug, clap::Args)]
+pub struct AddArgs {
+	pub path: PathBuf,
+}
+
+#[derive(Debug, clap::Args)]
+pub struct RemoveArgs {
+	pub path: PathBuf,
+}
+
+#[derive(Debug, clap::Args)]
+
+pub struct ListArgs;
+
+impl Cli {
+	pub async fn command_watch(&self, args: Args) -> tg::Result<()> {
+		match args.command {
+			Command::Add(args) => self.command_watch_add(args).await,
+			Command::List(args) => self.command_watch_list(args).await,
+			Command::Remove(args) => self.command_watch_remove(args).await,
+		}
+	}
+
+	async fn command_watch_add(&self, args: AddArgs) -> tg::Result<()> {
+		let client = &self.client().await?;
+		let path = args
+			.path
+			.canonicalize()
+			.map_err(|source| tg::error!(!source, "failed to canonicalize the path"))?;
+		let path = path.try_into()?;
+		tg::Artifact::check_in(client, &path, true)
+			.await
+			.map_err(|source| tg::error!(!source, %path, "failed to add watch"))?;
+		Ok(())
+	}
+
+	async fn command_watch_list(&self, _args: ListArgs) -> tg::Result<()> {
+		let client = self.client().await?;
+		let paths = client
+			.get_watches()
+			.await
+			.map_err(|source| tg::error!(!source, "failed to get watches"))?;
+		for path in paths {
+			eprintln!("{path}");
+		}
+		Ok(())
+	}
+
+	async fn command_watch_remove(&self, args: RemoveArgs) -> tg::Result<()> {
+		let client = self.client().await?;
+		let path = args.path.try_into()?;
+		client
+			.remove_watch(&path)
+			.await
+			.map_err(|source| tg::error!(!source, %path, "failed to remove watch"))?;
+		Ok(())
+	}
+}

packages/cli/src/commands/watch.rs

@@ -1,3 +1,11 @@
+//! # Configuring the tangram CLI and server
+//!
+//! Tangram can be configured by a global config file located at $HOME/.config/config.json or by passing the `--config <path>` option to the `tg` command line before any subcommand, for example
+//!
+//! ```sh
+//! # Run the server using a config file.
+//! tg --config config.json server run
+//! ```
 use serde_with::serde_as;
 use std::path::PathBuf;
 use tangram_client as tg;
@@ -6,57 +14,75 @@ use url::Url;
 #[serde_as]
 #[derive(Clone, Debug, Default, serde::Serialize, serde::Deserialize)]
 pub struct Config {
+	/// Advanced configuration options.
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub advanced: Option<Advanced>,
 
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub autoenv: Option<Autoenv>,
 
+	/// Configure the server's build options.
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub build: Option<Build>,
 
+	/// Configure the server's database options.
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub database: Option<Database>,
 
+	/// Configure the server's file system monitoring options.
+	#[serde(default, skip_serializing_if = "Option::is_none")]
+	pub file_system_monitor: Option<FileSystemMonitor>,
+
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub messenger: Option<Messenger>,
 
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub oauth: Option<Oauth>,
 
+	/// Configure the server's path. Default = `$HOME/.tangram`.
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub path: Option<PathBuf>,
 
+	/// A list of remote servers that this server can push and pull objects/builds from.
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub remotes: Option<Vec<Remote>>,
 
+	/// Server and CLI tracing options.
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub tracing: Option<Tracing>,
 
+	/// The URL of the server, if serving over tcp.
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub url: Option<Url>,
 
+	/// Configurate the virtual file system.
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub vfs: Option<Vfs>,
 }
 
 #[derive(Clone, Debug, Default, serde::Serialize, serde::Deserialize)]
 pub struct Advanced {
+	/// Configure how errors are displayed in the CLI.
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub error_trace_options: Option<tg::error::TraceOptions>,
 
+	/// Configure the number of file descriptors available in the client.
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub file_descriptor_limit: Option<u64>,
 
+	/// Configure the number of file descriptors available in the server.
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub file_descriptor_semaphore_size: Option<usize>,
 
+	/// Toggle whether temp directories are preserved or deleted after builds. Default = false.
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub preserve_temp_directories: Option<bool>,
 
+	/// Toggle whether tokio-console support is enabled. Default = false.
 	#[serde(default, skip_serializing_if = "std::ops::Not::not")]
 	pub tokio_console: bool,
 
+	/// Toggle whether log messages are printed to the server's stderr. Default = false.
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub write_build_logs_to_stderr: Option<bool>,
 }
@@ -69,7 +95,7 @@ pub struct Autoenv {
 
 #[derive(Clone, Debug, Default, serde::Serialize, serde::Deserialize)]
 pub struct Build {
-	/// Enable builds.
+	/// Toggle whether builds are enabled on this server.
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub enable: Option<bool>,
 
@@ -85,6 +111,12 @@ pub enum Database {
 	Postgres(PostgresDatabase),
 }
 
+#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
+pub struct FileSystemMonitor {
+	/// Toggle whether the file system monitor is enabled. Default = true.
+	pub enable: bool,
+}
+
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 pub struct SqliteDatabase {
 	/// The maximum number of connections.
@@ -149,8 +181,11 @@ pub struct RemoteBuild {
 
 #[derive(Clone, Debug, Default, serde::Serialize, serde::Deserialize)]
 pub struct Tracing {
+	/// The filter applied to tracing messages.
 	#[serde(default, skip_serializing_if = "String::is_empty")]
 	pub filter: String,
+
+	/// The display format of tracing messages.
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub format: Option<TracingFormat>,
 }
@@ -190,6 +225,7 @@ impl std::str::FromStr for TracingFormat {
 
 #[derive(Clone, Debug, Default, serde::Serialize, serde::Deserialize)]
 pub struct Vfs {
+	/// Toggle whether the VFS is enabled. When the VFS is disabled, checkouts will be made onto local disk. Default = true.
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub enable: Option<bool>,
 }

packages/cli/src/config.rs

@@ -76,6 +76,7 @@ pub enum Command {
 	Tree(self::commands::tree::Args),
 	Update(self::commands::package::UpdateArgs),
 	Upgrade(self::commands::upgrade::Args),
+	Watch(self::commands::watch::Args),
 }
 
 fn default_path() -> PathBuf {
@@ -212,6 +213,7 @@ impl Cli {
 			Command::Tree(args) => self.command_tree(args).boxed(),
 			Command::Update(args) => self.command_package_update(args).boxed(),
 			Command::Upgrade(args) => self.command_upgrade(args).boxed(),
+			Command::Watch(args) => self.command_watch(args).boxed(),
 		}
 		.await
 	}

packages/cli/src/main.rs

@@ -75,6 +75,7 @@ pub enum Data {
 #[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
 pub struct CheckInArg {
 	pub path: crate::Path,
+	pub watch: bool,
 }
 
 #[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
@@ -123,8 +124,11 @@ impl Artifact {
 }
 
 impl Artifact {
-	pub async fn check_in(tg: &impl Handle, path: &crate::Path) -> tg::Result<Self> {
-		let arg = CheckInArg { path: path.clone() };
+	pub async fn check_in(tg: &impl Handle, path: &crate::Path, watch: bool) -> tg::Result<Self> {
+		let arg = CheckInArg {
+			path: path.clone(),
+			watch,
+		};
 		let output = tg.check_in_artifact(arg).await?;
 		let artifact = Self::with_id(output.id);
 		Ok(artifact)

packages/client/src/artifact.rs

@@ -261,7 +261,7 @@ impl Blob {
 
 		// Check in the extracted artifact.
 		let path = path.try_into()?;
-		let artifact = Artifact::check_in(tg, &path)
+		let artifact = Artifact::check_in(tg, &path, false)
 			.await
 			.map_err(|source| error!(!source, "failed to check in the extracted archive"))?;
 

packages/client/src/blob.rs

@@ -308,4 +308,8 @@ pub trait Handle: Clone + Unpin + Send + Sync + 'static {
 		&self,
 		token: &str,
 	) -> impl Future<Output = tg::Result<Option<tg::User>>> + Send;
+
+	fn get_watches(&self) -> impl Future<Output = tg::Result<Vec<tg::Path>>> + Send;
+
+	fn remove_watch(&self, path: &tg::Path) -> impl Future<Output = tg::Result<()>> + Send;
 }

packages/client/src/handle.rs

@@ -781,4 +781,12 @@ impl Handle for Client {
 	async fn get_user_for_token(&self, token: &str) -> tg::Result<Option<tg::User>> {
 		self.get_user_for_token(token).await
 	}
+
+	async fn get_watches(&self) -> tg::Result<Vec<tg::Path>> {
+		self.get_watches().await
+	}
+
+	async fn remove_watch(&self, path: &tg::Path) -> tg::Result<()> {
+		self.remove_watch(path).await
+	}
 }