Mar 28, 2025 · Mar 21, 2025 · Mar 24, 2025 · Mar 24, 2025
diff --git a/Coder-Desktop/VPNLib/FileSync/FileSyncDaemon.swift b/Coder-Desktop/VPNLib/FileSync/FileSyncDaemon.swift

 @MainActor
 publicclassMutagenDaemon:FileSyncDaemon{
 privateletlogger=Logger(subsystem:Bundle.main.bundleIdentifier!, category:"mutagen")
 letlogger=Logger(subsystem:Bundle.main.bundleIdentifier!, category:"mutagen")

 @Publishedpublicvarstate:DaemonState=.stopped{
        didSet{
 privateletmutagenDaemonSocket:URL

    // Non-nil when the daemon is running
 varclient:DaemonClient?
 privatevargroup:MultiThreadedEventLoopGroup?
 privatevarchannel:GRPCChannel?
 privatevarclient:DaemonClient?

    // Protect start & stop transitions against re-entrancy
 privatelettransition=AsyncSemaphore(value:1)
 )
            client=DaemonClient(
                mgmt:Daemon_DaemonAsyncClient(channel: channel!),
                sync:Synchronization_SynchronizationAsyncClient(channel: channel!)
                sync:Synchronization_SynchronizationAsyncClient(channel: channel!),
                prompt:Prompting_PromptingAsyncClient(channel: channel!)
 )
            logger.info(
 "Successfully connected to mutagen daemon, socket:\(self.mutagenDaemonSocket.path, privacy:.public)"
 structDaemonClient{
 letmgmt:Daemon_DaemonAsyncClient
 letsync:Synchronization_SynchronizationAsyncClient
 letprompt:Prompting_PromptingAsyncClient
 }

 publicenumDaemonState{
 case connectionFailure(Error)
 case terminatedUnexpectedly
 case grpcFailure(Error)
 case invalidGrpcResponse(String)
 case unexpectedStreamClosure

 publicvardescription:String{
 switchself{
 "The daemon must be started first"
 caselet.grpcFailure(error):
 "Failed to communicate with daemon:\(error)"
 caselet.invalidGrpcResponse(response):
 "Invalid gRPC response:\(response)"
 case.unexpectedStreamClosure:
 "Unexpected stream closure"
 }
 }

diff --git a/Coder-Desktop/VPNLib/FileSync/FileSyncPrompting.swift b/Coder-Desktop/VPNLib/FileSync/FileSyncPrompting.swift
 import GRPC

 extensionMutagenDaemon{
 typealiasPromptStream=GRPCAsyncBidirectionalStreamingCall<Prompting_HostRequest,Prompting_HostResponse>

 func host(allowPrompts:Bool=true)asyncthrows(DaemonError)->(PromptStream, identifier:String){
 letstream= client!.prompt.makeHostCall()

 do{
 tryawait stream.requestStream.send(.with{ reqin req.allowPrompts= allowPrompts})
 }catch{
 throw.grpcFailure(error)
 }

        // We can't make call `makeAsyncIterator` more than once
        // (as a for-loop would do implicitly)
 variter= stream.responseStream.makeAsyncIterator()

 letinitResp:Prompting_HostResponse?
 do{
            initResp=tryawait iter.next()
 }catch{
 throw.grpcFailure(error)
 }
 guardlet initRespelse{
 throw.unexpectedStreamClosure
 }
 try initResp.ensureValid(first:true, allowPrompts: allowPrompts)

 Task.detached(priority:.background){
 do{
 whilelet msg=tryawait iter.next(){
 try msg.ensureValid(first:false, allowPrompts: allowPrompts)
 varreply:Prompting_HostRequest=.init()
 if msg.isPrompt{
                        // Handle SSH key prompts
 if msg.message.contains("yes/no/[fingerprint]"){
                            reply.response="yes"
 }
                        // Any other messages that require a non-empty response will
                        // cause the create op to fail, showing an error. This is ok for now.
 }
 tryawait stream.requestStream.send(reply)
 }
 }catchlet errorasGRPCStatuswhere error.code==.cancelled{
 return
 } catch{
 self.logger.critical("Prompt stream failed:\(error)")
 }
 }
 return(stream, identifier: initResp.identifier)
 }
 }
diff --git a/Coder-Desktop/VPNLib/FileSync/MutagenConvert.swift b/Coder-Desktop/VPNLib/FileSync/MutagenConvert.swift
 func humanReadableBytes(_ bytes:UInt64)->String{
 ByteCountFormatter().string(fromByteCount:Int64(bytes))
 }

 extensionPrompting_HostResponse{
 func ensureValid(first:Bool, allowPrompts:Bool)throws(DaemonError){
 if first{
 if identifier.isEmpty{
 throw.invalidGrpcResponse("empty prompter identifier")
 }
 if isPrompt{
 throw.invalidGrpcResponse("unexpected message type specification")
 }
 if !message.isEmpty{
 throw.invalidGrpcResponse("unexpected message")
 }
 }else{
 if !identifier.isEmpty{
 throw.invalidGrpcResponse("unexpected prompter identifier")
 }
 if isPrompt, !allowPrompts{
 throw.invalidGrpcResponse("disallowed prompt message type")
 }
 }
 }
 }
Original file line number	Diff line number	Diff line change
Expand Up		@@ -19,7 +19,7 @@ public protocol FileSyncDaemon: ObservableObject {

		@MainActor
		publicclassMutagenDaemon:FileSyncDaemon{
		privateletlogger=Logger(subsystem:Bundle.main.bundleIdentifier!, category:"mutagen")
		letlogger=Logger(subsystem:Bundle.main.bundleIdentifier!, category:"mutagen")

		@Publishedpublicvarstate:DaemonState=.stopped{
		didSet{
Expand All		@@ -42,9 +42,9 @@ public class MutagenDaemon: FileSyncDaemon {
		privateletmutagenDaemonSocket:URL

		// Non-nil when the daemon is running
		varclient:DaemonClient?
Copy link MemberAuthor ethanndicksonMar 24, 2025 Choose a reason for hiding this comment The reason will be displayed to describe this comment to others.Learn more. Making this`internal` instead of`private` so we can split the daemon class up into multiple extensions, across multiple files.
		privatevargroup:MultiThreadedEventLoopGroup?
		privatevarchannel:GRPCChannel?
		privatevarclient:DaemonClient?

		// Protect start & stop transitions against re-entrancy
		privatelettransition=AsyncSemaphore(value:1)
Expand DownExpand Up		@@ -171,7 +171,8 @@ public class MutagenDaemon: FileSyncDaemon {
		)
		client=DaemonClient(
		mgmt:Daemon_DaemonAsyncClient(channel: channel!),
		sync:Synchronization_SynchronizationAsyncClient(channel: channel!)
		sync:Synchronization_SynchronizationAsyncClient(channel: channel!),
		prompt:Prompting_PromptingAsyncClient(channel: channel!)
		)
		logger.info(
		"Successfully connected to mutagen daemon, socket:\(self.mutagenDaemonSocket.path, privacy:.public)"
Expand DownExpand Up		@@ -301,6 +302,7 @@ public class MutagenDaemon: FileSyncDaemon {
		structDaemonClient{
		letmgmt:Daemon_DaemonAsyncClient
		letsync:Synchronization_SynchronizationAsyncClient
		letprompt:Prompting_PromptingAsyncClient
		}

		publicenumDaemonState{
Expand DownExpand Up		@@ -342,6 +344,8 @@ public enum DaemonError: Error {
		case connectionFailure(Error)
		case terminatedUnexpectedly
		case grpcFailure(Error)
		case invalidGrpcResponse(String)
		case unexpectedStreamClosure

		publicvardescription:String{
		switchself{
Expand All		@@ -355,6 +359,10 @@ public enum DaemonError: Error {
		"The daemon must be started first"
		caselet.grpcFailure(error):
		"Failed to communicate with daemon:\(error)"
		caselet.invalidGrpcResponse(response):
		"Invalid gRPC response:\(response)"
		case.unexpectedStreamClosure:
		"Unexpected stream closure"
		}
		}

Expand Down
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1,53 @@
		import GRPC

		extensionMutagenDaemon{
		typealiasPromptStream=GRPCAsyncBidirectionalStreamingCall<Prompting_HostRequest,Prompting_HostResponse>

		func host(allowPrompts:Bool=true)asyncthrows(DaemonError)->(PromptStream, identifier:String){
		letstream= client!.prompt.makeHostCall()

		do{
		tryawait stream.requestStream.send(.with{ reqin req.allowPrompts= allowPrompts})
		}catch{
		throw.grpcFailure(error)
		}

		// We can't make call `makeAsyncIterator` more than once
		// (as a for-loop would do implicitly)
		variter= stream.responseStream.makeAsyncIterator()

		letinitResp:Prompting_HostResponse?
		do{
		initResp=tryawait iter.next()
		}catch{
		throw.grpcFailure(error)
		}
		guardlet initRespelse{
		throw.unexpectedStreamClosure
		}
		try initResp.ensureValid(first:true, allowPrompts: allowPrompts)

		Task.detached(priority:.background){
Copy link MemberAuthor ethanndicksonMar 24, 2025 Choose a reason for hiding this comment The reason will be displayed to describe this comment to others.Learn more. This block is self-contained, and we're currently doing a lot on the main actor already. I was previously running into issues on startup with the VPN code getting starved by file sync code.
		do{
		whilelet msg=tryawait iter.next(){
		try msg.ensureValid(first:false, allowPrompts: allowPrompts)
		varreply:Prompting_HostRequest=.init()
		if msg.isPrompt{
		// Handle SSH key prompts
		if msg.message.contains("yes/no/[fingerprint]"){
		reply.response="yes"
		}
		// Any other messages that require a non-empty response will
		// cause the create op to fail, showing an error. This is ok for now.
		}
		tryawait stream.requestStream.send(reply)
		}
		}catchlet errorasGRPCStatuswhere error.code==.cancelled{
		return
		} catch{
		self.logger.critical("Prompt stream failed:\(error)")
ethanndickson marked this conversation as resolved. Show resolvedHide resolved
		}
		}
		return(stream, identifier: initResp.identifier)
		}
		}
Original file line number	Diff line number	Diff line change
Expand Up		@@ -57,3 +57,26 @@ func accumulateErrors(from state: Synchronization_State) -> [FileSyncError] {
		func humanReadableBytes(_ bytes:UInt64)->String{
		ByteCountFormatter().string(fromByteCount:Int64(bytes))
		}

		extensionPrompting_HostResponse{
		func ensureValid(first:Bool, allowPrompts:Bool)throws(DaemonError){
		if first{
		if identifier.isEmpty{
		throw.invalidGrpcResponse("empty prompter identifier")
		}
		if isPrompt{
		throw.invalidGrpcResponse("unexpected message type specification")
		}
		if !message.isEmpty{
		throw.invalidGrpcResponse("unexpected message")
		}
		}else{
		if !identifier.isEmpty{
		throw.invalidGrpcResponse("unexpected prompter identifier")
		}
		if isPrompt, !allowPrompts{
		throw.invalidGrpcResponse("disallowed prompt message type")
		}
		}
		}
		}