gRPC follow ups

ldk-server-client/src/client.rs

@@ -53,7 +53,11 @@ use ldk_server_grpc::endpoints::{
 	UNIFIED_SEND_PATH, UPDATE_CHANNEL_CONFIG_PATH, VERIFY_SIGNATURE_PATH,
 };
 use ldk_server_grpc::events::EventEnvelope;
-use ldk_server_grpc::grpc::{decode_grpc_body, encode_grpc_frame, percent_decode};
+use ldk_server_grpc::grpc::{
+	decode_grpc_body, encode_grpc_frame, percent_decode, GRPC_STATUS_FAILED_PRECONDITION,
+	GRPC_STATUS_INTERNAL, GRPC_STATUS_INVALID_ARGUMENT, GRPC_STATUS_OK,
+	GRPC_STATUS_UNAUTHENTICATED, GRPC_STATUS_UNAVAILABLE,
+};
 use prost::Message;
 use reqwest::{header::HeaderMap, Certificate, Client};
 use rustls::{ClientConfig, RootCertStore};
@@ -506,18 +510,18 @@ impl LdkServerClient {
 /// Map a gRPC status code to an LdkServerError.
 fn grpc_code_to_error(code: u32, message: String) -> LdkServerError {
 	match code {
-		3 => LdkServerError::new(InvalidRequestError, message), // INVALID_ARGUMENT
-		9 => LdkServerError::new(LightningError, message),      // FAILED_PRECONDITION
-		13 => LdkServerError::new(InternalServerError, message), // INTERNAL
-		14 => LdkServerError::new(
+		GRPC_STATUS_INVALID_ARGUMENT => LdkServerError::new(InvalidRequestError, message),
+		GRPC_STATUS_FAILED_PRECONDITION => LdkServerError::new(LightningError, message),
+		GRPC_STATUS_INTERNAL => LdkServerError::new(InternalServerError, message),
+		GRPC_STATUS_UNAVAILABLE => LdkServerError::new(
 			InternalError,
 			if message.is_empty() {
 				"gRPC stream became unavailable".to_string()
 			} else {
 				format!("gRPC stream became unavailable: {message}")
 			},
 		),
-		16 => LdkServerError::new(AuthError, message), // UNAUTHENTICATED
+		GRPC_STATUS_UNAUTHENTICATED => LdkServerError::new(AuthError, message),
 		_ => LdkServerError::new(
 			InternalError,
 			if message.is_empty() {
@@ -531,7 +535,7 @@ fn grpc_code_to_error(code: u32, message: String) -> LdkServerError {
 
 fn grpc_error_from_headers(headers: &HeaderMap) -> Option<LdkServerError> {
 	let code = headers.get("grpc-status")?.to_str().ok()?.parse::<u32>().ok()?;
-	if code == 0 {
+	if code == GRPC_STATUS_OK {
 		return None;
 	}
 
@@ -680,7 +684,7 @@ mod tests {
 
 	#[test]
 	fn test_grpc_code_to_error_marks_unavailable_streams() {
-		let err = grpc_code_to_error(14, "server shutting down".to_string());
+		let err = grpc_code_to_error(GRPC_STATUS_UNAVAILABLE, "server shutting down".to_string());
 		assert_eq!(err.error_code, InternalError);
 		assert_eq!(err.message, "gRPC stream became unavailable: server shutting down");
 	}
@@ -710,10 +714,10 @@ mod tests {
 	#[test]
 	fn test_grpc_code_to_error_all_known_codes() {
 		let cases = [
-			(3u32, InvalidRequestError, "msg"),
-			(16, AuthError, "msg"),
-			(9, LightningError, "msg"),
-			(13, InternalServerError, "msg"),
+			(GRPC_STATUS_INVALID_ARGUMENT, InvalidRequestError, "msg"),
+			(GRPC_STATUS_UNAUTHENTICATED, AuthError, "msg"),
+			(GRPC_STATUS_FAILED_PRECONDITION, LightningError, "msg"),
+			(GRPC_STATUS_INTERNAL, InternalServerError, "msg"),
 		];
 		for (code, expected_error_code, msg) in cases {
 			let err = grpc_code_to_error(code, msg.to_string());

ldk-server-grpc/src/grpc.rs

@@ -15,6 +15,7 @@
 use bytes::{BufMut, Bytes, BytesMut};
 
 // gRPC status codes (a subset — only those we use).
+pub const GRPC_STATUS_OK: u32 = 0;
 pub const GRPC_STATUS_INVALID_ARGUMENT: u32 = 3;
 pub const GRPC_STATUS_DEADLINE_EXCEEDED: u32 = 4;
 pub const GRPC_STATUS_FAILED_PRECONDITION: u32 = 9;
@@ -160,7 +161,8 @@ impl http_body::Body for GrpcBody {
 /// Build trailers for a successful gRPC response.
 fn ok_trailers() -> http::HeaderMap {
 	let mut trailers = http::HeaderMap::with_capacity(1);
-	trailers.insert("grpc-status", http::HeaderValue::from_static("0"));
+	trailers
+		.insert("grpc-status", http::HeaderValue::from_str(&GRPC_STATUS_OK.to_string()).unwrap());
 	trailers
 }
 

ldk-server/src/service.rs

@@ -369,6 +369,16 @@ impl Service<Request<Incoming>> for NodeService {
 					tokio::spawn(async move {
 						loop {
 							tokio::select! {
+								biased;
+								_ = shutdown_rx.changed() => {
+									let _ = tx
+										.send(Err(GrpcStatus::new(
+											GRPC_STATUS_UNAVAILABLE,
+											"server shutting down",
+										)))
+										.await;
+									break;
+								},
 								result = rx.recv() => {
 									match result {
 										Ok(event) => {
@@ -391,15 +401,6 @@ impl Service<Request<Incoming>> for NodeService {
 										},
 									}
 								},
-								_ = shutdown_rx.changed() => {
-									let _ = tx
-										.send(Err(GrpcStatus::new(
-											GRPC_STATUS_UNAVAILABLE,
-											"server shutting down",
-										)))
-										.await;
-									break;
-								},
 							}
 						}
 					});