use std::collections::VecDeque;
use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
use async_stream::stream;
use bytes::Bytes;
use crate::LixError;
use crate::Value;
use super::http::{
ProtocolHttp, ProtocolHttpRequest, ProtocolHttpResponse, ProtocolHttpStream, StreamCancel,
};
use super::wire::{SERVER_CLOSED_CODE, SERVER_PROTOCOL_VERSION, SESSION_GONE_CODE};
use super::{ProtocolExecuteOptions, open_protocol_client};
#[test]
fn connection_locator_maps_to_the_targeted_protocol_root() {
assert_eq!(
super::normalize_protocol_base_url(
"https://example.test/lix/01936f4e-7b6c-7c3d-8f9a-123456789abc"
)
.expect("canonical locator"),
"https://example.test/lix/v1/01936f4e-7b6c-7c3d-8f9a-123456789abc/"
);
}
#[test]
fn connection_locator_rejects_raw_protocol_and_noncanonical_ids() {
for invalid in [
"https://example.test/lix/v1/01936f4e-7b6c-7c3d-8f9a-123456789abc",
"https://example.test/lix/01936F4E-7B6C-7C3D-8F9A-123456789ABC",
"https://example.test/lix/not-a-uuid",
"https://example.test/prefix/lix/01936f4e-7b6c-7c3d-8f9a-123456789abc",
"http://example.test/lix/01936f4e-7b6c-7c3d-8f9a-123456789abc",
"https://user@example.test/lix/01936f4e-7b6c-7c3d-8f9a-123456789abc",
"https://example.test/lix/01936f4e-7b6c-7c3d-8f9a-123456789abc?token=secret",
"https://example.test/lix/01936f4e-7b6c-7c3d-8f9a-123456789abc/",
] {
assert!(
super::normalize_protocol_base_url(invalid).is_err(),
"accepted invalid locator: {invalid}"
);
}
assert!(super::normalize_protocol_base_url(
"http://127.0.0.1:3000/lix/01936f4e-7b6c-7c3d-8f9a-123456789abc"
)
.is_ok());
assert!(super::normalize_protocol_base_url(
"http://[::1]:3000/lix/01936f4e-7b6c-7c3d-8f9a-123456789abc"
)
.is_ok());
}
#[derive(Clone, Default)]
struct ScriptHttp {
requests: Arc<Mutex<Vec<ProtocolHttpRequest>>>,
outcomes: Arc<Mutex<VecDeque<ScriptOutcome>>>,
stream_cancellations: Arc<AtomicUsize>,
sleeps: Arc<Mutex<Vec<Duration>>>,
}
enum ScriptOutcome {
Json {
status: u16,
body: serde_json::Value,
},
Empty {
status: u16,
},
Stream {
status: u16,
headers: Vec<(String, String)>,
chunks: Vec<Bytes>,
},
}
impl ScriptHttp {
fn push_json(&self, status: u16, body: serde_json::Value) {
self.outcomes
.lock()
.expect("script outcomes")
.push_back(ScriptOutcome::Json { status, body });
}
fn push_empty(&self, status: u16) {
self.outcomes
.lock()
.expect("script outcomes")
.push_back(ScriptOutcome::Empty { status });
}
fn push_stream(&self, status: u16, body: &str) {
self.outcomes
.lock()
.expect("script outcomes")
.push_back(ScriptOutcome::Stream {
status,
headers: vec![("content-type".to_owned(), "text/event-stream".to_owned())],
chunks: vec![Bytes::from(body.to_owned())],
});
}
fn requests(&self) -> Vec<ProtocolHttpRequest> {
self.requests.lock().expect("script requests").clone()
}
}
impl ProtocolHttp for ScriptHttp {
async fn request(
&self,
request: ProtocolHttpRequest,
) -> Result<ProtocolHttpResponse, LixError> {
self.requests
.lock()
.expect("script requests")
.push(request.clone());
match self.outcomes.lock().expect("script outcomes").pop_front() {
Some(ScriptOutcome::Json { status, body }) => Ok(ProtocolHttpResponse {
status,
headers: vec![("content-type".to_owned(), "application/json".to_owned())],
body: Bytes::from(serde_json::to_vec(&body).expect("script json")),
}),
Some(ScriptOutcome::Empty { status }) => Ok(ProtocolHttpResponse {
status,
headers: Vec::new(),
body: Bytes::new(),
}),
Some(ScriptOutcome::Stream { .. }) => Err(LixError::new(
"LIX_SERVER_PROTOCOL_ERROR",
"scripted stream used as a finite request",
)),
None => Err(LixError::new(
"LIX_REMOTE_UNAVAILABLE",
"no scripted response remaining",
)),
}
}
async fn request_stream(
&self,
request: ProtocolHttpRequest,
) -> Result<ProtocolHttpStream, LixError> {
self.requests
.lock()
.expect("script requests")
.push(request.clone());
match self.outcomes.lock().expect("script outcomes").pop_front() {
Some(ScriptOutcome::Stream {
status,
headers,
chunks,
}) => {
let cancellations = self.stream_cancellations.clone();
let cancel: StreamCancel = Arc::new(move || {
cancellations.fetch_add(1, Ordering::SeqCst);
});
Ok(ProtocolHttpStream {
status,
headers,
body: Box::pin(stream! {
for chunk in chunks {
yield Ok(chunk);
}
}),
cancel,
})
}
Some(ScriptOutcome::Json { status, body }) => {
let cancellations = self.stream_cancellations.clone();
let cancel: StreamCancel = Arc::new(move || {
cancellations.fetch_add(1, Ordering::SeqCst);
});
Ok(ProtocolHttpStream {
status,
headers: vec![("content-type".to_owned(), "application/json".to_owned())],
body: Box::pin(stream! {
yield Ok(Bytes::from(serde_json::to_vec(&body).expect("script json")));
}),
cancel,
})
}
Some(other) => {
self.outcomes
.lock()
.expect("script outcomes")
.push_front(other);
Err(LixError::new(
"LIX_REMOTE_UNAVAILABLE",
"no scripted stream remaining",
))
}
None => Err(LixError::new(
"LIX_REMOTE_UNAVAILABLE",
"no scripted stream remaining",
)),
}
}
async fn sleep(&self, duration: Duration) {
self.sleeps.lock().unwrap().push(duration);
}
fn spawn(&self, fut: Pin<Box<dyn Future<Output = ()> + Send>>) {
tokio::spawn(fut);
}
}
fn handshake(session_id: &str, branch_id: &str) -> serde_json::Value {
handshake_with_account(
session_id,
branch_id,
"00000000-0000-7000-8000-000000000002",
)
}
fn handshake_with_account(
session_id: &str,
branch_id: &str,
account_id: &str,
) -> serde_json::Value {
serde_json::json!({
"protocolVersion": SERVER_PROTOCOL_VERSION,
"activeBranchId": branch_id,
"activeAccountId": account_id,
"sessionId": session_id,
})
}
fn protocol_error(code: &str, status: u16) -> serde_json::Value {
serde_json::json!({
"error": {
"code": code,
"message": code,
},
"httpStatus": status,
})
}
fn execute_ok() -> serde_json::Value {
serde_json::json!({
"columns": [{ "name": "n", "type": "integer" }],
"rows": [[{ "kind": "int", "value": 1 }]],
"rowsAffected": 0,
"notices": [],
})
}
fn sse_next(subscription_id: &str) -> String {
format!(
"event: next\ndata: {}\n\n",
serde_json::json!({
"subscriptionId": subscription_id,
"sequence": 0,
"mutationSequence": 1,
"result": execute_ok(),
})
)
}
fn sse_error(code: &str) -> String {
format!(
"event: error\ndata: {}\n\n",
serde_json::json!({
"error": { "code": code, "message": code },
})
)
}
#[tokio::test]
async fn child_sessions_inherit_branch_and_retain_snapshot_export() {
let http = ScriptHttp::default();
http.push_json(200, handshake("parent", "branch-a"));
http.push_json(200, handshake("child", "branch-a"));
http.push_json(200, handshake("grandchild", "branch-b"));
http.push_stream(200, "snapshot bytes");
let parent = open_protocol_client(
http.clone(),
"https://lix.test/lix/01936f4e-7b6c-7c3d-8f9a-123456789abc",
None,
)
.await
.expect("parent");
let child = parent
.open_another_session(None, None)
.await
.expect("child");
let grandchild = child
.open_another_session(Some("branch-b".to_owned()), None)
.await
.expect("grandchild");
assert_eq!(child.active_branch_id().await.expect("branch"), "branch-a");
assert_eq!(parent.session_id().as_deref(), Some("parent"));
assert_eq!(grandchild.session_id().as_deref(), Some("grandchild"));
let mut snapshot = grandchild.export_snapshot().await.expect("export");
assert_eq!(
snapshot.next().await.expect("chunk"),
Some(Bytes::from_static(b"snapshot bytes"))
);
assert_eq!(snapshot.next().await.expect("end"), None);
assert_eq!(http.stream_cancellations.load(Ordering::SeqCst), 1);
let requests = http.requests();
assert!(requests[1].url.ends_with("/?activeBranchId=branch-a"));
assert!(requests[2].url.ends_with("/?activeBranchId=branch-b"));
assert!(
requests[3]
.url
.ends_with("/lix/v1/01936f4e-7b6c-7c3d-8f9a-123456789abc/snapshot")
);
assert_eq!(
requests[3].header("accept"),
Some("application/vnd.lix.snapshot")
);
assert_eq!(requests[3].header("lix-session-id"), None);
}
#[tokio::test]
async fn child_session_rejects_account_override_without_opening_a_session() {
let http = ScriptHttp::default();
http.push_json(200, handshake("parent", "branch-a"));
let parent = open_protocol_client(
http.clone(),
"https://lix.test/lix/01936f4e-7b6c-7c3d-8f9a-123456789abc",
None,
)
.await
.expect("parent");
let error = parent
.open_another_session(None, Some("another-account".to_owned()))
.await
.expect_err("account override");
assert_eq!(error.code, LixError::CODE_INVALID_PARAM);
assert_eq!(http.requests().len(), 1);
}
#[tokio::test]
async fn child_session_closes_new_session_if_authentication_changes() {
let http = ScriptHttp::default();
http.push_json(200, handshake("parent", "branch-a"));
http.push_json(
200,
handshake_with_account("child", "branch-a", "another-account"),
);
http.push_empty(204);
let parent = open_protocol_client(
http.clone(),
"https://lix.test/lix/01936f4e-7b6c-7c3d-8f9a-123456789abc",
None,
)
.await
.expect("parent");
let error = parent
.open_another_session(None, None)
.await
.expect_err("authentication change");
assert_eq!(error.code, LixError::CODE_INVALID_PARAM);
let requests = http.requests();
assert_eq!(requests[2].method, "DELETE");
assert_eq!(requests[2].header("lix-session-id"), Some("child"));
assert_eq!(
parent
.active_branch_id()
.await
.expect("parent remains usable"),
"branch-a"
);
}
#[tokio::test]
async fn snapshot_errors_use_protocol_error_envelopes_and_cancel_the_stream() {
let http = ScriptHttp::default();
http.push_json(200, handshake("parent", "branch-a"));
http.push_json(
403,
serde_json::json!({"error": {
"code": "LIX_ACCESS_DENIED", "message": "denied", "hint": "request access",
"details": {"scope": "snapshot"}
}}),
);
let parent = open_protocol_client(
http.clone(),
"https://lix.test/lix/01936f4e-7b6c-7c3d-8f9a-123456789abc",
None,
)
.await
.expect("parent");
let error = parent.export_snapshot().await.err().expect("denied");
assert_eq!(error.code, "LIX_ACCESS_DENIED");
assert_eq!(error.hint.as_deref(), Some("request access"));
assert_eq!(error.details.as_ref().expect("details")["httpStatus"], 403);
assert_eq!(http.stream_cancellations.load(Ordering::SeqCst), 1);
}
#[tokio::test]
async fn snapshot_cancel_and_drop_release_the_stream_once() {
let http = ScriptHttp::default();
http.push_json(200, handshake("parent", "branch-a"));
http.push_stream(200, "snapshot bytes");
http.push_stream(200, "snapshot bytes");
let parent = open_protocol_client(
http.clone(),
"https://lix.test/lix/01936f4e-7b6c-7c3d-8f9a-123456789abc",
None,
)
.await
.expect("parent");
let mut snapshot = parent.export_snapshot().await.expect("export");
snapshot.cancel();
snapshot.cancel();
assert_eq!(snapshot.next().await.expect("cancelled"), None);
drop(snapshot);
assert_eq!(http.stream_cancellations.load(Ordering::SeqCst), 1);
drop(parent.export_snapshot().await.expect("export"));
assert_eq!(http.stream_cancellations.load(Ordering::SeqCst), 2);
}
#[tokio::test]
async fn execute_recovers_once_on_session_gone_and_pins_the_last_branch() {
let http = ScriptHttp::default();
http.push_json(200, handshake("session-1", "branch-a"));
http.push_json(410, protocol_error(SESSION_GONE_CODE, 410));
http.push_json(200, handshake("session-2", "branch-a"));
http.push_json(200, execute_ok());
http.push_empty(204);
let client = open_protocol_client(http.clone(), "https://lix.test/lix/01936f4e-7b6c-7c3d-8f9a-123456789abc", None)
.await
.expect("open");
assert_eq!(client.active_branch_id().await.expect("branch"), "branch-a");
client
.execute("SELECT 1", &[], None)
.await
.expect("recovered execute");
client.close().await.expect("close");
let requests = http.requests();
assert_eq!(requests[0].method, "GET");
assert!(!requests[0].url.contains("activeBranchId="));
assert!(requests[0].header("Lix-Session-Id").is_none());
assert_eq!(requests[1].method, "POST");
assert!(requests[1].url.ends_with("/execute"));
assert_eq!(requests[1].header("Lix-Session-Id"), Some("session-1"));
assert_eq!(requests[2].method, "GET");
assert!(requests[2].url.contains("activeBranchId=branch-a"));
assert!(requests[2].header("Lix-Session-Id").is_none());
assert_eq!(requests[3].method, "POST");
assert_eq!(requests[3].header("Lix-Session-Id"), Some("session-2"));
}
#[tokio::test]
async fn execute_recovers_once_on_server_closed() {
let http = ScriptHttp::default();
http.push_json(200, handshake("session-1", "main"));
http.push_json(503, protocol_error(SERVER_CLOSED_CODE, 503));
http.push_json(200, handshake("session-2", "main"));
http.push_json(200, execute_ok());
http.push_empty(204);
let client = open_protocol_client(
http.clone(),
"https://lix.test/lix/01936f4e-7b6c-7c3d-8f9a-123456789abc",
None,
)
.await
.expect("open");
client
.execute(
"SELECT 1",
&[Value::Integer(1)],
Some(ProtocolExecuteOptions {
origin_key: None,
max_auto_commit_retries: Some(0),
idempotency_key: Some("retry-1".to_owned()),
}),
)
.await
.expect("recovered execute");
for request in http
.requests()
.iter()
.filter(|request| request.method == "POST")
{
let body: serde_json::Value = serde_json::from_slice(request.body.as_deref().unwrap()).unwrap();
assert_eq!(body["options"]["maxAutoCommitRetries"], 0);
}
client.close().await.expect("close");
assert!(
http.requests()
.iter()
.filter(|request| request.method == "GET")
.count()
>= 2
);
}
#[tokio::test]
async fn execute_second_session_gone_fails_without_another_handshake() {
let http = ScriptHttp::default();
http.push_json(200, handshake("session-1", "main"));
http.push_json(410, protocol_error(SESSION_GONE_CODE, 410));
http.push_json(200, handshake("session-2", "main"));
http.push_json(410, protocol_error(SESSION_GONE_CODE, 410));
http.push_empty(204);
let client = open_protocol_client(http.clone(), "https://lix.test/lix/01936f4e-7b6c-7c3d-8f9a-123456789abc", None)
.await
.expect("open");
let error = client
.execute("SELECT 1", &[], None)
.await
.expect_err("second gone must fail");
assert_eq!(error.code, SESSION_GONE_CODE);
client.close().await.expect("close");
let handshakes = http
.requests()
.into_iter()
.filter(|request| request.method == "GET")
.count();
assert_eq!(handshakes, 2);
}
#[tokio::test]
async fn session_recovery_rejects_an_authenticated_account_change() {
let http = ScriptHttp::default();
http.push_json(200, handshake_with_account("session-1", "main", "account-a"));
http.push_json(410, protocol_error(SESSION_GONE_CODE, 410));
http.push_json(200, handshake_with_account("session-2", "main", "account-b"));
let client = open_protocol_client(
http.clone(),
"https://lix.test/lix/01936f4e-7b6c-7c3d-8f9a-123456789abc",
None,
)
.await
.expect("open");
let error = client
.execute("SELECT 1", &[], None)
.await
.expect_err("recovery must not change authenticated account");
assert_eq!(error.code, "LIX_SERVER_PROTOCOL_ERROR");
assert!(error.message.contains("activeAccountId"));
let poisoned = client
.execute("SELECT 1", &[], None)
.await
.expect_err("an account-mismatched client must remain unusable");
assert_eq!(poisoned.code, "LIX_SERVER_PROTOCOL_ERROR");
assert_eq!(
http.requests()
.iter()
.filter(|request| request.url.ends_with("/execute"))
.count(),
1,
"the failed recovery must not retry under the new account"
);
}
#[tokio::test]
async fn open_handshake_can_pin_an_initial_branch() {
let http = ScriptHttp::default();
http.push_json(200, handshake("session-1", "draft / one"));
http.push_empty(204);
let client = open_protocol_client(
http.clone(),
"https://lix.test/lix/01936f4e-7b6c-7c3d-8f9a-123456789abc",
Some("draft / one".to_owned()),
)
.await
.expect("open");
assert_eq!(
client.active_account_id().await.expect("account"),
"00000000-0000-7000-8000-000000000002"
);
client.close().await.expect("close");
let handshake_url = &http.requests()[0].url;
assert!(handshake_url.contains("/lix/v1/01936f4e-7b6c-7c3d-8f9a-123456789abc/"));
assert!(
handshake_url.contains("activeBranchId=draft+%2F+one") || handshake_url.contains("draft")
);
assert!(http.requests()[0].header("Lix-Session-Id").is_none());
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[ignore = "spawned observe success path is covered by SESSION_GONE reconnect-loop tests"]
async fn observe_recovers_once_on_session_gone() {
let http = ScriptHttp::default();
http.push_json(200, handshake("session-1", "main"));
http.push_json(410, protocol_error(SESSION_GONE_CODE, 410));
http.push_json(200, handshake("session-2", "main"));
http.push_stream(200, &sse_next("observe-1"));
http.push_json(200, execute_ok());
http.push_stream(200, "event: message\ndata: \n\n");
http.push_empty(204);
let client = open_protocol_client(http.clone(), "https://lix.test/lix/01936f4e-7b6c-7c3d-8f9a-123456789abc", None)
.await
.expect("open");
let events = client
.observe("SELECT 1", Vec::new())
.await
.expect("observe");
let event = tokio::time::timeout(Duration::from_secs(2), events.next())
.await
.expect("observe next timed out")
.expect("observe next");
assert!(event.is_some());
events.close();
client.close().await.expect("close");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn observe_second_session_gone_fails_without_a_reconnect_loop() {
let http = ScriptHttp::default();
http.push_json(200, handshake("session-1", "main"));
http.push_json(410, protocol_error(SESSION_GONE_CODE, 410));
http.push_json(200, handshake("session-2", "main"));
http.push_json(410, protocol_error(SESSION_GONE_CODE, 410));
http.push_empty(204);
let client = open_protocol_client(http.clone(), "https://lix.test/lix/01936f4e-7b6c-7c3d-8f9a-123456789abc", None)
.await
.expect("open");
let events = client
.observe("SELECT 1", Vec::new())
.await
.expect("observe");
let error = tokio::time::timeout(Duration::from_secs(2), events.next())
.await
.expect("observe next timed out")
.expect_err("second gone must fail");
assert_eq!(error.code, SESSION_GONE_CODE);
events.close();
client.close().await.expect("close");
let observe_opens = http
.requests()
.into_iter()
.filter(|request| request.url.contains("/observe/multiplex"))
.count();
assert!(
observe_opens <= 2,
"observe reconnect looped: {observe_opens} opens"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn observe_error_event_recovers_once_then_fails() {
let http = ScriptHttp::default();
http.push_json(200, handshake("session-1", "main"));
http.push_stream(200, &sse_error(SESSION_GONE_CODE));
http.push_json(200, handshake("session-2", "main"));
http.push_stream(200, &sse_error(SESSION_GONE_CODE));
http.push_empty(204);
let client = open_protocol_client(http.clone(), "https://lix.test/lix/01936f4e-7b6c-7c3d-8f9a-123456789abc", None)
.await
.expect("open");
let events = client
.observe("SELECT 1", Vec::new())
.await
.expect("observe");
let error = tokio::time::timeout(Duration::from_secs(2), events.next())
.await
.expect("observe next timed out")
.expect_err("second gone must fail");
assert_eq!(error.code, SESSION_GONE_CODE);
events.close();
client.close().await.expect("close");
}
#[cfg(feature = "server-protocol")]
#[tokio::test]
async fn observe_client_request_passes_server_version_gate_and_streams_initial_result() {
use super::observe::ObserveTransport as _;
use crate::server_protocol::{
SERVER_PROTOCOL_VERSION_HEADER, ServerProtocolBody, ServerProtocolContext,
};
use http_body_util::BodyExt as _;
let server = crate::open_lix()
.with_storage(crate::Memory::new())
.serve()
.with_embedded_lix_id()
.await
.expect("serve observation fixture");
let handshake_response = server
.handle(
http::Request::builder()
.uri(format!("/lix/v1/{}", server.lix_id()))
.header(SERVER_PROTOCOL_VERSION_HEADER, SERVER_PROTOCOL_VERSION)
.body(ServerProtocolBody::empty())
.expect("handshake request"),
ServerProtocolContext::anonymous(),
)
.await;
assert_eq!(handshake_response.status(), http::StatusCode::OK);
let handshake_body = handshake_response
.into_body()
.collect()
.await
.expect("handshake body")
.to_bytes();
let http = ScriptHttp::default();
http.push_json(
200,
serde_json::from_slice(&handshake_body).expect("handshake JSON"),
);
http.push_stream(200, "");
let client = open_protocol_client(
http.clone(),
format!("https://lix.test/lix/{}", server.lix_id()),
None,
)
.await
.expect("open client with real server session");
let captured_stream = client
.core
.open_observe_stream(vec![super::wire::MultiplexObserveSubscription {
id: "version-gate".to_owned(),
sql: "SELECT 42 AS answer".to_owned(),
params: Vec::new(),
}])
.await
.expect("capture actual client observe request");
(captured_stream.cancel)();
let request = http.requests().pop().expect("captured observe request");
assert_eq!(
request.header(SERVER_PROTOCOL_VERSION_HEADER),
Some(SERVER_PROTOCOL_VERSION.to_string().as_str())
);
let replay = |include_version: bool| {
let mut builder = http::Request::builder()
.method(request.method.as_str())
.uri(&request.url);
for (name, value) in &request.headers {
if include_version || !name.eq_ignore_ascii_case(SERVER_PROTOCOL_VERSION_HEADER) {
builder = builder.header(name, value);
}
}
builder
.body(ServerProtocolBody::from(
request.body.clone().expect("observe body"),
))
.expect("replayed observe request")
};
let rejected = server
.handle(replay(false), ServerProtocolContext::anonymous())
.await;
assert_eq!(rejected.status(), http::StatusCode::UPGRADE_REQUIRED);
let response = server
.handle(replay(true), ServerProtocolContext::anonymous())
.await;
assert_eq!(response.status(), http::StatusCode::OK);
let mut body = response.into_body();
let frame = tokio::time::timeout(Duration::from_secs(2), body.frame())
.await
.expect("initial result timeout")
.expect("initial frame")
.expect("valid frame")
.into_data()
.expect("SSE data");
let event = std::str::from_utf8(&frame).expect("UTF-8 SSE");
assert!(
event.contains("event: next"),
"expected initial result, got {event}"
);
assert!(
event.contains("version-gate") && event.contains("42"),
"unexpected result: {event}"
);
drop(body);
server.close().await.expect("close server");
}
#[tokio::test]
async fn opening_awaits_typed_migration_without_a_total_deadline() {
let http = ScriptHttp::default();
for index in 0..40 {
let code = if index % 2 == 0 {
"LIX_REPOSITORY_MIGRATING"
} else {
"LIX_ERROR_MIGRATING"
};
http.push_json(
503,
serde_json::json!({"error":{"code":code,"message":"migration in progress"}}),
);
}
http.push_json(200, handshake("ready", "main"));
let client = open_protocol_client(
http.clone(),
"https://lix.test/lix/01936f4e-7b6c-7c3d-8f9a-123456789abc",
None,
)
.await
.unwrap();
assert_eq!(http.requests().len(), 41);
assert!(
http.requests()
.iter()
.all(|request| request.method == "GET")
);
assert_eq!(
*http.sleeps.lock().unwrap(),
vec![Duration::from_secs(1); 40]
);
http.push_json(503, serde_json::json!({"error":{"code":"LIX_REPOSITORY_MIGRATING","message":"migration in progress"}}));
let error = client
.execute("INSERT INTO example VALUES (1)", &[], None)
.await
.unwrap_err();
assert_eq!(error.code, "LIX_REPOSITORY_MIGRATING");
assert_eq!(http.requests().len(), 42);
assert_eq!(http.sleeps.lock().unwrap().len(), 40);
}
#[tokio::test]
async fn opening_does_not_delay_healthy_admission_or_retry_terminal_errors() {
let http = ScriptHttp::default();
http.push_json(200, handshake("ready", "main"));
let _client = open_protocol_client(
http.clone(),
"https://lix.test/lix/01936f4e-7b6c-7c3d-8f9a-123456789abc",
None,
)
.await
.unwrap();
assert_eq!(http.requests().len(), 1);
assert!(http.sleeps.lock().unwrap().is_empty());
for (status, code) in [
(503, "LIX_ERROR_UNAVAILABLE"),
(401, "LIX_REPOSITORY_MIGRATING"),
(426, "LIX_SYNC_PROTOCOL_MISMATCH"),
] {
let http = ScriptHttp::default();
http.push_json(503, serde_json::json!({"error":{"code":"LIX_REPOSITORY_MIGRATING","message":"migration in progress"}}));
http.push_json(
status,
serde_json::json!({"error":{"code":code,"message":"terminal"}}),
);
let error = open_protocol_client(
http.clone(),
"https://lix.test/lix/01936f4e-7b6c-7c3d-8f9a-123456789abc",
None,
)
.await
.unwrap_err();
assert_eq!(error.code, code);
assert_eq!(http.requests().len(), 2);
assert_eq!(*http.sleeps.lock().unwrap(), vec![Duration::from_secs(1)]);
}
}