Allow farmer to start quickly in dev environment and shut down quickl…

…y once node RPC disconnects (#2535)

crates/subspace-farmer/src/single_disk_farm/farming.rs

@@ -105,6 +105,9 @@ pub enum FarmingError {
         /// Lower-level error
         error: node_client::Error,
     },
+    /// Slot info notification stream ended
+    #[error("Slot info notification stream ended")]
+    SlotNotificationStreamEnded,
     /// Low-level auditing error
     #[error("Low-level auditing error: {0}")]
     LowLevelAuditing(#[from] AuditingError),
@@ -150,6 +153,7 @@ impl FarmingError {
             FarmingError::Io(_) => "Io",
             FarmingError::FailedToCreateThreadPool(_) => "FailedToCreateThreadPool",
             FarmingError::Decoded(_) => "Decoded",
+            FarmingError::SlotNotificationStreamEnded => "SlotNotificationStreamEnded",
         }
     }
 
@@ -163,6 +167,7 @@ impl FarmingError {
             FarmingError::Io(_) => true,
             FarmingError::FailedToCreateThreadPool(_) => true,
             FarmingError::Decoded(error) => error.is_fatal,
+            FarmingError::SlotNotificationStreamEnded => true,
         }
     }
 }
@@ -193,7 +198,7 @@ where
         }
     }
 
-    Ok(())
+    Err(FarmingError::SlotNotificationStreamEnded)
 }
 
 /// Plot audit options

crates/subspace-service/src/lib.rs

@@ -853,6 +853,11 @@ where
             ))
         });
 
+        if !config.base.network.force_synced {
+            // Start with DSN sync in this case
+            pause_sync.store(true, Ordering::Release);
+        }
+
         let (observer, worker) = sync_from_dsn::create_observer_and_worker(
             segment_headers_store.clone(),
             Arc::clone(&network_service),

crates/subspace-service/src/sync_from_dsn.rs

@@ -249,8 +249,7 @@ where
         .saturating_sub(chain_constants.confirmation_depth_k().into());
     let segment_header_downloader = SegmentHeaderDownloader::new(node);
 
-    // Node starts as offline, we'll wait for it to go online shrtly after
-    let mut initial_pause_sync = Some(pause_sync.swap(true, Ordering::AcqRel));
+    let mut initial_pause_sync = Some(pause_sync.load(Ordering::Acquire));
     while let Some(reason) = notifications.next().await {
         let prev_pause_sync = pause_sync.swap(true, Ordering::AcqRel);