feat(job): disable auto-subscription for job activity and enhance job context management

This commit removes the auto-subscription feature for job activity, streamlining the job registration process. Additionally, it enhances the JobManager by adding fields for job name and action context, allowing for better tracking and management of job-related information. These changes improve the overall clarity and efficiency of job handling within the system.

core/src/infra/job/manager.rs

@@ -43,6 +43,8 @@ struct RunningJob {
 	status_tx: watch::Sender<JobStatus>,
 	latest_progress: Arc<Mutex<Option<Progress>>>,
 	persistence_complete_rx: Option<tokio::sync::oneshot::Receiver<()>>,
+	job_name: String,
+	action_context: Option<crate::infra::action::context::ActionContext>,
 }
 
 impl JobManager {
@@ -363,6 +365,8 @@ impl JobManager {
 						status_tx: status_tx.clone(),
 						latest_progress,
 						persistence_complete_rx: Some(persistence_complete_rx),
+						job_name: job_name.to_string(),
+						action_context: action_context.clone(),
 					},
 				);
 
@@ -785,6 +789,8 @@ impl JobManager {
 						status_tx: status_tx.clone(),
 						latest_progress: latest_progress.clone(),
 						persistence_complete_rx: Some(persistence_complete_rx),
+						job_name: J::NAME.to_string(),
+						action_context: action_context.clone(),
 					},
 				);
 
@@ -1060,8 +1066,23 @@ impl JobManager {
 					0.0
 				};
 
-			// Get job data from database for complete info
-			let (job_name, action_type, action_context) =
+			// Get job data from in-memory struct (for non-persisted jobs) or database
+			let (job_name, action_type, action_context) = if let Some(ctx) = &running_job.action_context {
+				// Use in-memory action_context (for ephemeral volume jobs)
+				let action_context_info = ActionContextInfo {
+					action_type: ctx.action_type.clone(),
+					initiated_at: ctx.initiated_at,
+					initiated_by: ctx.initiated_by.clone(),
+					action_input: ctx.action_input.clone().into(),
+					context: ctx.context.clone().into(),
+				};
+				(
+					running_job.job_name.clone(),
+					Some(ctx.action_type.clone()),
+					Some(action_context_info),
+				)
+			} else {
+				// Fall back to database query for persisted jobs
 				match database::jobs::Entity::find_by_id(job_id.0.to_string())
 					.one(self.db.conn())
 					.await?
@@ -1086,8 +1107,9 @@ impl JobManager {
 						};
 						(db_job.name, db_job.action_type, action_context)
 					}
-					None => (format!("Job {}", job_id.0), None, None),
-				};
+					None => (running_job.job_name.clone(), None, None),
+				}
+			};
 
 			all_jobs.push(JobInfo {
 				id: job_id.0,
@@ -1481,6 +1503,13 @@ impl JobManager {
 								// Clone latest_progress for monitoring task before moving into RunningJob
 								let latest_progress_for_monitor = latest_progress.clone();
 
+								// Deserialize action context from database if available
+								let action_context = if let Some(context_data) = &job_record.action_context {
+									rmp_serde::from_slice::<crate::infra::action::context::ActionContext>(context_data).ok()
+								} else {
+									None
+								};
+
 								self.running_jobs.write().await.insert(
 									job_id,
 									RunningJob {
@@ -1489,6 +1518,8 @@ impl JobManager {
 										status_tx: status_tx.clone(),
 										latest_progress,
 										persistence_complete_rx: Some(persistence_complete_rx),
+										job_name: job_record.name.clone(),
+										action_context,
 									},
 								);
 
@@ -1779,12 +1810,19 @@ impl JobManager {
 					)));
 				}
 
-				Some((job_record.name.clone(), job_record.state.clone()))
+				// Deserialize action context from database if available
+				let action_context = if let Some(context_data) = &job_record.action_context {
+					rmp_serde::from_slice::<crate::infra::action::context::ActionContext>(context_data).ok()
+				} else {
+					None
+				};
+
+				Some((job_record.name.clone(), job_record.state.clone(), action_context))
 			}
 		};
 
 		// If job was not in memory, recreate and dispatch it
-		if let Some((job_name, job_state)) = job_info {
+		if let Some((job_name, job_state, action_context)) = job_info {
 			// Deserialize job from binary data
 			info!(
 				"RESUME_STATE_LOAD: Job {} loading {} bytes of state from database (manual resume)",
@@ -1917,6 +1955,8 @@ impl JobManager {
 					status_tx: status_tx.clone(),
 					latest_progress,
 					persistence_complete_rx: Some(persistence_complete_rx),
+					job_name: job_name.clone(),
+					action_context,
 				},
 			);
 

core/src/lib.rs

@@ -697,38 +697,7 @@ async fn register_default_protocol_handlers(
 			.await;
 	}
 
-	// Set up job activity client for auto-subscription
-	let job_activity_client = service::network::JobActivityClient::new(
-		networking
-			.endpoint()
-			.cloned()
-			.ok_or("Endpoint not initialized")?,
-		networking.active_connections(),
-		context.remote_job_cache.clone(),
-		networking.device_registry(),
-	);
-
-	// Auto-subscribe to job activity from connected devices
-	let mut event_subscriber = networking.subscribe_events();
-
-	tokio::spawn(async move {
-		while let Ok(event) = event_subscriber.recv().await {
-			if let service::network::NetworkEvent::ConnectionEstablished { device_id, .. } = event {
-				if let Err(e) = job_activity_client
-					.subscribe_to_device(device_id, None)
-					.await
-				{
-					tracing::error!(
-						"Auto-subscribe to job activity failed for device {}: {}",
-						device_id,
-						e
-					);
-				} else {
-					tracing::info!("Auto-subscribed to job activity from device {}", device_id);
-				}
-			}
-		}
-	});
+	// Job activity auto-subscription disabled
 
 	// Brief delay to ensure protocol handlers are fully initialized and background
 	// tasks have started before accepting connections. This prevents race conditions

core/src/ops/indexing/ephemeral/index.rs

@@ -136,6 +136,17 @@ impl EphemeralIndex {
 		path: PathBuf,
 		uuid: Uuid,
 		metadata: EntryMetadata,
+	) -> std::io::Result<Option<ContentKind>> {
+		let registry = FileTypeRegistry::default();
+		self.add_entry_with_registry(path, uuid, metadata, &registry)
+	}
+
+	fn add_entry_with_registry(
+		&mut self,
+		path: PathBuf,
+		uuid: Uuid,
+		metadata: EntryMetadata,
+		registry: &FileTypeRegistry,
 	) -> std::io::Result<Option<ContentKind>> {
 		if self.path_index.contains_key(&path) {
 			tracing::trace!("Skipping duplicate entry: {}", path.display());
@@ -184,7 +195,6 @@ impl EphemeralIndex {
 		}
 
 		let content_kind = if metadata.kind == EntryKind::File {
-			let registry = FileTypeRegistry::default();
 			registry.identify_by_extension(&path)
 		} else if metadata.kind == EntryKind::Directory {
 			ContentKind::Unknown
@@ -210,8 +220,11 @@ impl EphemeralIndex {
 	) -> std::io::Result<Vec<Option<ContentKind>>> {
 		let mut results = Vec::with_capacity(entries.len());
 
+		// Create registry once for entire batch instead of per-file
+		let registry = FileTypeRegistry::default();
+
 		for (path, uuid, metadata) in entries {
-			let result = self.add_entry(path, uuid, metadata)?;
+			let result = self.add_entry_with_registry(path, uuid, metadata, &registry)?;
 			results.push(result);
 		}
 

core/src/ops/indexing/job.rs

@@ -382,6 +382,7 @@ impl IndexerJob {
 							ephemeral_index,
 							root_path,
 							volume_backend.as_ref(),
+							self.config.is_volume_indexing,
 						)
 						.await?;
 					} else {
@@ -881,6 +882,7 @@ impl IndexerJob {
 		ephemeral_index: Arc<RwLock<EphemeralIndex>>,
 		root_path: &Path,
 		_volume_backend: Option<&Arc<dyn crate::volume::VolumeBackend>>,
+		is_volume_indexing: bool,
 	) -> JobResult<()> {
 		use super::database_storage::EntryMetadata;
 		use super::state::EntryKind as StateEntryKind;
@@ -916,16 +918,18 @@ impl IndexerJob {
 			ctx.progress(Progress::generic(indexer_progress.to_generic_progress()));
 
 			// Convert DirEntry to (PathBuf, Uuid, EntryMetadata) tuples
+			// Keep the UUIDs so we don't need to query them back
 			let entries_with_metadata: Vec<(PathBuf, Uuid, EntryMetadata)> = batch
 				.iter()
 				.map(|entry| {
+					let uuid = Uuid::new_v4();
 					let metadata = EntryMetadata {
 						path: entry.path.clone(),
 						kind: entry.kind,
 						size: entry.size,
 						modified: entry.modified,
 						accessed: None,
-						created: None, // DirEntry doesn't have created time
+						created: None,
 						inode: entry.inode,
 						permissions: None,
 						is_hidden: entry
@@ -935,101 +939,116 @@ impl IndexerJob {
 							.map(|n| n.starts_with('.'))
 							.unwrap_or(false),
 					};
-					(entry.path.clone(), Uuid::new_v4(), metadata)
+					(entry.path.clone(), uuid, metadata)
 				})
 				.collect();
 
-			// Batch add to index (single write lock acquisition)
-			let content_kinds = {
-				let mut index = ephemeral_index.write().await;
-				index.add_entries_batch(entries_with_metadata)?
-			};
+			// Batch add to index - use spawn_blocking for CPU-intensive work
+			// This allows the task to be interrupted even during blocking operations
+			let index_clone = ephemeral_index.clone();
+			let entries_clone = entries_with_metadata.clone();
+			let content_kinds = tokio::task::spawn_blocking(move || {
+				let rt = tokio::runtime::Handle::current();
+				let mut index = rt.block_on(index_clone.write());
+				index.add_entries_batch(entries_clone)
+			})
+			.await
+			.map_err(|e| JobError::execution(format!("Failed to add entries to index: {}", e)))??;
 
-			// Collect files for ResourceChangedBatch event
-			let mut files_for_event = Vec::new();
+			// Build UUID lookup map (avoids 1.4M lock acquisitions)
+			let uuid_map: std::collections::HashMap<PathBuf, Uuid> = entries_with_metadata
+				.iter()
+				.map(|(path, uuid, _)| (path.clone(), *uuid))
+				.collect();
 
-			for (entry, content_kind_opt) in batch.iter().zip(content_kinds.iter()) {
-				if let Some(content_kind) = content_kind_opt {
-					// Skip hidden files from events
-					let is_hidden = entry
-						.path
-						.file_name()
-						.and_then(|n| n.to_str())
-						.map(|n| n.starts_with('.'))
-						.unwrap_or(false);
+			// Only emit file events for directory browsing, not volume indexing
+			// Volume indexing only needs job progress events (emitted above)
+			// Directory browsing needs ResourceChangedBatch events to populate UI
+			if !is_volume_indexing {
+				// Build event files using the UUID map (no lock acquisitions)
+				let files_for_event: Vec<File> = batch
+					.iter()
+					.zip(content_kinds.iter())
+					.filter_map(|(entry, content_kind_opt)| {
+						let content_kind = (*content_kind_opt)?;
 
-					if !is_hidden {
-						// Build File for event
-						let uuid = {
-							let index = ephemeral_index.read().await;
-							index.get_entry_uuid(&entry.path)
-						};
+						// Skip hidden files from events
+						let is_hidden = entry
+							.path
+							.file_name()
+							.and_then(|n| n.to_str())
+							.map(|n| n.starts_with('.'))
+							.unwrap_or(false);
 
-						if let Some(uuid) = uuid {
-							use chrono::{DateTime, Utc};
-
-							let file = File {
-								id: uuid,
-								sd_path: crate::domain::addressing::SdPath::local(entry.path.clone()),
-								kind: match entry.kind {
-									StateEntryKind::File => DomainEntryKind::File,
-									StateEntryKind::Directory => DomainEntryKind::Directory,
-									StateEntryKind::Symlink => DomainEntryKind::Symlink,
-								},
-								name: entry
-									.path
-									.file_name()
-									.unwrap_or_default()
-									.to_string_lossy()
-									.to_string(),
-								extension: entry
-									.path
-									.extension()
-									.and_then(|e| e.to_str())
-									.map(String::from),
-								size: entry.size,
-								content_identity: None,
-								alternate_paths: vec![],
-								tags: vec![],
-								sidecars: vec![],
-								image_media_data: None,
-								video_media_data: None,
-								audio_media_data: None,
-								created_at: Utc::now(), // Unknown, use current time
-								modified_at: entry
-									.modified
-									.and_then(|t| DateTime::from_timestamp(
-										t.duration_since(std::time::UNIX_EPOCH)
-											.ok()?
-											.as_secs() as i64,
-										0,
-									))
-									.unwrap_or_else(Utc::now),
-								accessed_at: None,
-								content_kind: *content_kind,
-								is_local: true,
-								duration_seconds: None,
-							};
-							files_for_event.push(file);
+						if is_hidden {
+							return None;
 						}
-					}
-				}
-			}
 
-			// Emit single ResourceChangedBatch event for entire batch
-			if !files_for_event.is_empty() {
-				event_bus.emit(crate::infra::event::Event::ResourceChangedBatch {
-					resource_type: "file".to_string(),
-					resources: serde_json::to_value(&files_for_event).unwrap_or_default(),
-					metadata: Some(crate::infra::event::ResourceMetadata {
-						no_merge_fields: vec![],
-						alternate_ids: vec![],
-						affected_paths: files_for_event
-							.iter()
-							.map(|f| f.sd_path.clone())
-							.collect(),
-					}),
-				});
+						// Get UUID from our map (no lock acquisition needed!)
+						let uuid = *uuid_map.get(&entry.path)?;
+
+						use chrono::{DateTime, Utc};
+
+						Some(File {
+							id: uuid,
+							sd_path: crate::domain::addressing::SdPath::local(entry.path.clone()),
+							kind: match entry.kind {
+								StateEntryKind::File => DomainEntryKind::File,
+								StateEntryKind::Directory => DomainEntryKind::Directory,
+								StateEntryKind::Symlink => DomainEntryKind::Symlink,
+							},
+							name: entry
+								.path
+								.file_name()
+								.unwrap_or_default()
+								.to_string_lossy()
+								.to_string(),
+							extension: entry
+								.path
+								.extension()
+								.and_then(|e| e.to_str())
+								.map(String::from),
+							size: entry.size,
+							content_identity: None,
+							alternate_paths: vec![],
+							tags: vec![],
+							sidecars: vec![],
+							image_media_data: None,
+							video_media_data: None,
+							audio_media_data: None,
+							created_at: Utc::now(),
+							modified_at: entry
+								.modified
+								.and_then(|t| DateTime::from_timestamp(
+									t.duration_since(std::time::UNIX_EPOCH)
+										.ok()?
+										.as_secs() as i64,
+									0,
+								))
+								.unwrap_or_else(Utc::now),
+							accessed_at: None,
+							content_kind,
+							is_local: true,
+							duration_seconds: None,
+						})
+					})
+					.collect();
+
+				// Emit single ResourceChangedBatch event for entire batch
+				if !files_for_event.is_empty() {
+					event_bus.emit(crate::infra::event::Event::ResourceChangedBatch {
+						resource_type: "file".to_string(),
+						resources: serde_json::to_value(&files_for_event).unwrap_or_default(),
+						metadata: Some(crate::infra::event::ResourceMetadata {
+							no_merge_fields: vec![],
+							alternate_ids: vec![],
+							affected_paths: files_for_event
+								.iter()
+								.map(|f| f.sd_path.clone())
+								.collect(),
+						}),
+					});
+				}
 			}
 		}
 

packages/interface/src/components/JobManager/types.ts

@@ -103,6 +103,13 @@ export function getJobDisplayName(job: JobListItem): string {
         return "Extracting Media";
       case "indexing.start":
         return "Indexing Location";
+      case "volumes.index": {
+        const volumeName = job.action_context?.context?.volume_name;
+        if (volumeName && typeof volumeName === 'string') {
+          return `Indexing ${volumeName}`;
+        }
+        return "Indexing Volume";
+      }
       default: {
         // Capitalize and format action type
         return action_type