diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml index 9d78a79..0237755 100644 --- a/.github/workflows/integration.yml +++ b/.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 diff --git a/docker-compose.yml b/docker-compose.yml index 5e41f6f..f23baa5 100644 --- a/docker-compose.yml +++ b/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: diff --git a/trogon-eventstore/src/commands.rs b/trogon-eventstore/src/commands.rs index 35542ad..c05ae31 100644 --- a/trogon-eventstore/src/commands.rs +++ b/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, diff --git a/trogon-eventstore/src/operations/gossip.rs b/trogon-eventstore/src/operations/gossip.rs index 1065c4b..b59e44d 100644 --- a/trogon-eventstore/src/operations/gossip.rs +++ b/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, Box> { - 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::().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, -} - #[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, - 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 { diff --git a/trogon-eventstore/src/operations/mod.rs b/trogon-eventstore/src/operations/mod.rs index 4dfc365..dafa27d 100644 --- a/trogon-eventstore/src/operations/mod.rs +++ b/trogon-eventstore/src/operations/mod.rs @@ -73,12 +73,9 @@ impl Client { pub async fn read_gossip(&self) -> crate::Result> { 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 { diff --git a/trogon-eventstore/src/types.rs b/trogon-eventstore/src/types.rs index 2f704d4..6e2fcad 100755 --- a/trogon-eventstore/src/types.rs +++ b/trogon-eventstore/src/types.rs @@ -1127,14 +1127,14 @@ pub enum SubscriptionEvent { #[derive(Debug)] pub struct CaughtUp { - pub date: DateTime, + pub timestamp: DateTime, pub stream_revision: Option, pub position: Option, } #[derive(Debug)] pub struct FellBehind { - pub date: DateTime, + pub timestamp: DateTime, pub stream_revision: Option, pub position: Option, } diff --git a/trogon-eventstore/tests/api/operations.rs b/trogon-eventstore/tests/api/operations.rs index 667c7a8..1e363fe 100644 --- a/trogon-eventstore/tests/api/operations.rs +++ b/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(()) } diff --git a/trogon-eventstore/tests/api/streams.rs b/trogon-eventstore/tests/api/streams.rs index 080e249..8d371fc 100644 --- a/trogon-eventstore/tests/api/streams.rs +++ b/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 }); 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(()) } diff --git a/trogon-eventstore/tests/images.rs b/trogon-eventstore/tests/images.rs index 8f38fb7..73a70e1 100644 --- a/trogon-eventstore/tests/images.rs +++ b/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![], } } diff --git a/trogon-eventstore/tests/integration.rs b/trogon-eventstore/tests/integration.rs index 0de0fb2..d1407c2 100644 --- a/trogon-eventstore/tests/integration.rs +++ b/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