fix(client): align with the server contract

Signed-off-by: Straw Hat Team Bot <61149376+sht-bot@users.noreply.github.com>

Author: yordis

.github/workflows/integration.yml

@@ -7,22 +7,22 @@ on:
         description: Server image registry
         required: true
         type: string
-        default: docker.io
+        default: ghcr.io
       server_repository:
         description: Server image repository
         required: true
         type: string
-        default: eventstore
+        default: trogonstack
       server_container:
         description: Server image name
         required: true
         type: string
-        default: eventstore
+        default: trogoneventstore
       server_version:
         description: Server image tag
         required: true
         type: string
-        default: latest
+        default: ci
 
 permissions:
   contents: read

docker-compose.yml

@@ -24,7 +24,7 @@ services:
       - volumes-provisioner
 
   esdb-node1: &template
-    image: ${ESDB_DOCKER_REGISTRY:-docker.io}/${ESDB_DOCKER_REPO:-eventstore}/${ESDB_DOCKER_CONTAINER:-eventstore}:${ESDB_DOCKER_CONTAINER_VERSION:-latest}
+    image: ${ESDB_DOCKER_REGISTRY:-ghcr.io}/${ESDB_DOCKER_REPO:-trogonstack}/${ESDB_DOCKER_CONTAINER:-trogoneventstore}:${ESDB_DOCKER_CONTAINER_VERSION:-ci}
     env_file:
       - vars.env
     environment:

trogon-eventstore/src/commands.rs

@@ -841,7 +841,7 @@ impl Subscription {
 
                                 streams::read_resp::Content::CaughtUp(args) => {
                                     let args = args.timestamp.map(|t| crate::CaughtUp {
-                                        date: timestamp_to_datetime(t),
+                                        timestamp: timestamp_to_datetime(t),
                                         stream_revision: args.stream_revision.map(|x| x as u64),
                                         position: args.position.map(|x| Position {
                                             commit: x.commit_position,
@@ -854,7 +854,7 @@ impl Subscription {
 
                                 streams::read_resp::Content::FellBehind(args) => {
                                     let args = args.timestamp.map(|t| crate::FellBehind {
-                                        date: timestamp_to_datetime(t),
+                                        timestamp: timestamp_to_datetime(t),
                                         stream_revision: args.stream_revision.map(|x| x as u64),
                                         position: args.position.map(|x| Position {
                                             commit: x.commit_position,

trogon-eventstore/src/operations/gossip.rs

@@ -1,9 +1,8 @@
+use crate::ClientSettings;
 use crate::event_store::client::gossip as wire;
 use crate::grpc::HyperClient;
-use crate::http::http_configure_auth;
 use crate::request::build_request_metadata;
 use crate::types::Endpoint;
-use crate::{ClientSettings, grpc};
 use serde::{Deserialize, Serialize};
 use tonic::{Request, Status};
 use uuid::Uuid;
@@ -71,57 +70,6 @@ pub async fn read(
     Ok(members)
 }
 
-pub(crate) async fn http_read(
-    setts: &ClientSettings,
-    handle: grpc::Handle,
-) -> Result<Vec<MemberInfo>, Box<dyn std::error::Error>> {
-    let client = reqwest::Client::builder()
-        .danger_accept_invalid_certs(!setts.tls_verify_cert)
-        .build()?;
-
-    let default_auth = setts
-        .default_user_name
-        .as_ref()
-        .map(|c| crate::Authentication::Basic(c.clone()));
-
-    let resp = http_configure_auth(
-        client.get(format!("{}/gossip", handle.url())),
-        default_auth.as_ref(),
-    )
-    .send()
-    .await?;
-
-    let gossip = resp.json::<Gossip>().await?;
-
-    Ok(gossip
-        .members
-        .into_iter()
-        .map(|i| MemberInfo {
-            instance_id: i.instance_id,
-            time_stamp: i.time_stamp.timestamp(),
-            state: i.state,
-            is_alive: i.is_alive,
-            http_end_point: Endpoint {
-                host: i.external_http_ip,
-                port: i.external_http_port as u32,
-            },
-            last_commit_position: i.last_commit_position,
-            writer_checkpoint: i.writer_checkpoint,
-            chaser_checkpoint: i.chaser_checkpoint,
-            epoch_position: i.epoch_position,
-            epoch_number: i.epoch_number,
-            epoch_id: i.epoch_id,
-            node_priority: i.node_priority,
-        })
-        .collect())
-}
-
-#[derive(Serialize, Deserialize, Debug)]
-#[serde(rename_all = "camelCase")]
-struct Gossip {
-    members: Vec<HttpMemberInfo>,
-}
-
 #[derive(Debug, Clone)]
 pub struct MemberInfo {
     pub instance_id: Uuid,
@@ -138,31 +86,6 @@ pub struct MemberInfo {
     pub node_priority: i64,
 }
 
-#[derive(Deserialize, Serialize, Debug)]
-#[serde(rename_all = "camelCase")]
-pub struct HttpMemberInfo {
-    pub instance_id: Uuid,
-    pub time_stamp: chrono::DateTime<chrono::Utc>,
-    pub state: VNodeState,
-    pub is_alive: bool,
-    pub internal_tcp_ip: String,
-    pub internal_tcp_port: u16,
-    pub internal_secure_tcp_port: u16,
-    pub external_tcp_ip: String,
-    pub external_secure_tcp_port: u16,
-    #[serde(rename = "httpEndPointIp")]
-    pub external_http_ip: String,
-    #[serde(rename = "httpEndPointPort")]
-    pub external_http_port: u16,
-    pub last_commit_position: i64,
-    pub writer_checkpoint: i64,
-    pub chaser_checkpoint: i64,
-    pub epoch_position: i64,
-    pub epoch_number: i64,
-    pub epoch_id: Uuid,
-    pub node_priority: i64,
-}
-
 #[derive(Copy, Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
 #[serde(rename_all = "PascalCase")]
 pub enum VNodeState {

trogon-eventstore/src/operations/mod.rs

@@ -73,12 +73,9 @@ impl Client {
     pub async fn read_gossip(&self) -> crate::Result<Vec<gossip::MemberInfo>> {
         let handle = self.inner.current_selected_node().await?;
 
-        // We currently use the http endpoint instead of the gRPC one because at that time
-        // 04-25-2022, the public gRPC endpoint doesn't return all the gossip info like current
-        // epoch and other checkpoints.
-        gossip::http_read(self.inner.connection_settings(), handle)
+        gossip::read(self.inner.connection_settings(), &handle.client, handle.uri)
             .await
-            .map_err(|e| crate::Error::IllegalStateError(e.to_string()))
+            .map_err(crate::Error::from_grpc)
     }
 
     pub async fn stats(&self, options: &StatsOptions) -> crate::Result<Stats> {

trogon-eventstore/src/types.rs

@@ -1127,14 +1127,14 @@ pub enum SubscriptionEvent {
 
 #[derive(Debug)]
 pub struct CaughtUp {
-    pub date: DateTime<Utc>,
+    pub timestamp: DateTime<Utc>,
     pub stream_revision: Option<u64>,
     pub position: Option<Position>,
 }
 
 #[derive(Debug)]
 pub struct FellBehind {
-    pub date: DateTime<Utc>,
+    pub timestamp: DateTime<Utc>,
     pub stream_revision: Option<u64>,
     pub position: Option<Position>,
 }

trogon-eventstore/tests/api/operations.rs

@@ -1,5 +1,6 @@
 use std::time::Duration;
 use tracing::debug;
+use trogon_eventstore::Credentials;
 use trogon_eventstore::operations;
 use trogon_eventstore::operations::StatsOptions;
 
@@ -186,12 +187,15 @@ async fn test_change_user_password(
         )
         .await?;
 
+    let options = operations::OperationalOptions::default()
+        .authenticated(Credentials::new(login.clone(), password.clone()));
+
     client
         .change_user_password(
             login.as_str(),
-            password,
+            password.as_str(),
             names.next().unwrap(),
-            &Default::default(),
+            &options,
         )
         .await?;
 
@@ -242,10 +246,7 @@ async fn test_op_restart_persistent_subscription_subsystem(
 }
 
 async fn test_scavenge(client: &operations::Client) -> trogon_eventstore::Result<()> {
-    let result = client.start_scavenge(1, 0, &Default::default()).await?;
-    let result = client.stop_scavenge(result.id(), &Default::default()).await;
-
-    assert!(result.is_ok());
+    client.start_scavenge(1, 0, &Default::default()).await?;
 
     Ok(())
 }

trogon-eventstore/tests/api/streams.rs

@@ -1,6 +1,5 @@
 use crate::common::{fresh_stream_id, generate_events};
 use chrono::{Datelike, Utc};
-use futures::channel::oneshot;
 use std::collections::HashMap;
 use std::time::Duration;
 use tracing::{debug, warn};
@@ -265,9 +264,7 @@ async fn test_subscription(client: &Client) -> eyre::Result<()> {
         .subscribe_to_stream(stream_id.as_str(), &options)
         .await;
 
-    let (tx, recv) = oneshot::channel();
-
-    tokio::spawn(async move {
+    let subscription = tokio::spawn(async move {
         let mut count = 0usize;
         let max = 6usize;
 
@@ -280,20 +277,20 @@ async fn test_subscription(client: &Client) -> eyre::Result<()> {
             }
         }
 
-        tx.send(count).unwrap();
-        Ok(()) as trogon_eventstore::Result<()>
+        Ok(count) as trogon_eventstore::Result<usize>
     });
 
     let _ = client
         .append_to_stream(stream_id, &Default::default(), events_after)
         .await?;
 
-    match tokio::time::timeout(Duration::from_secs(60), recv).await {
+    match tokio::time::timeout(Duration::from_secs(60), subscription).await {
         Ok(test_count) => {
+            let test_count = test_count??;
             assert_eq!(
-                test_count?, 6,
+                test_count, 6,
                 "We are testing proper state after catchup subscription: got {} expected {}.",
-                test_count?, 6
+                test_count, 6
             );
         }
 
@@ -328,25 +325,20 @@ async fn test_subscription_caughtup(client: &Client) -> trogon_eventstore::Resul
         .subscribe_to_stream(stream_id.clone(), &options)
         .await;
 
-    let (tx, recv) = oneshot::channel();
-
-    tokio::spawn(async move {
+    let caught_up = tokio::time::timeout(Duration::from_secs(60), async move {
         loop {
-            if let SubscriptionEvent::CaughtUp(_) = sub.next_subscription_event().await? {
-                break;
+            if let SubscriptionEvent::CaughtUp(caught_up) = sub.next_subscription_event().await? {
+                return Ok::<_, trogon_eventstore::Error>(caught_up);
             }
         }
+    })
+    .await
+    .expect("test_subscription_caughtup timed out")?
+    .expect("server did not provide caught-up context");
 
-        let _ = tx.send(());
-        Ok(()) as trogon_eventstore::Result<()>
-    });
-
-    if tokio::time::timeout(Duration::from_secs(60), recv)
-        .await
-        .is_err()
-    {
-        panic!("test_subscription_caughtup timed out!");
-    }
+    assert!(caught_up.timestamp <= Utc::now());
+    assert_eq!(caught_up.stream_revision, Some(9));
+    assert_eq!(caught_up.position, None);
 
     Ok(())
 }

trogon-eventstore/tests/images.rs

@@ -6,10 +6,10 @@ use testcontainers::{
     core::{ContainerPort, Mount, WaitFor},
 };
 
-const DEFAULT_REGISTRY: &str = "docker.io";
-const DEFAULT_REPO: &str = "eventstore";
-const DEFAULT_CONTAINER: &str = "eventstore";
-const DEFAULT_TAG: &str = "latest";
+const DEFAULT_REGISTRY: &str = "ghcr.io";
+const DEFAULT_REPO: &str = "trogonstack";
+const DEFAULT_CONTAINER: &str = "trogoneventstore";
+const DEFAULT_TAG: &str = "ci";
 
 #[derive(Debug, Clone)]
 pub struct EventStoreDB {
@@ -23,10 +23,6 @@ impl EventStoreDB {
     pub fn insecure_mode(mut self) -> Self {
         self.env_vars
             .insert("EVENTSTORE_INSECURE".to_string(), "true".to_string());
-        self.env_vars.insert(
-            "EVENTSTORE_ENABLE_ATOM_PUB_OVER_HTTP".to_string(),
-            "true".to_string(),
-        );
 
         self
     }
@@ -163,16 +159,10 @@ impl Default for EventStoreDB {
         let tag = option_env!("ESDB_DOCKER_CONTAINER_VERSION").unwrap_or(DEFAULT_TAG);
         let repo = option_env!("ESDB_DOCKER_REPO").unwrap_or(DEFAULT_REPO);
         let container = option_env!("ESDB_DOCKER_CONTAINER").unwrap_or(DEFAULT_CONTAINER);
-        let mut env_vars = HashMap::new();
-
-        env_vars.insert(
-            "EVENTSTORE_GOSSIP_ON_SINGLE_NODE".to_string(),
-            "true".to_string(),
-        );
         EventStoreDB {
             name: format!("{}/{}/{}", registry, repo, container),
             tag: tag.to_string(),
-            env_vars,
+            env_vars: HashMap::new(),
             mounts: vec![],
         }
     }

trogon-eventstore/tests/integration.rs

@@ -57,7 +57,7 @@ async fn wait_node_is_alive(
             match tokio::time::timeout(
                 std::time::Duration::from_secs(1),
                 client
-                    .get(format!("{}://localhost:{}/health/live", protocol, port))
+                    .get(format!("{}://localhost:{}/-/readiness", protocol, port))
                     .send(),
             )
             .await