Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions .github/workflows/integration.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion docker-compose.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
4 changes: 2 additions & 2 deletions trogon-eventstore/src/commands.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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,
Expand Down
79 changes: 1 addition & 78 deletions trogon-eventstore/src/operations/gossip.rs
Original file line number Diff line number Diff line change
@@ -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;
Expand Down Expand Up @@ -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,
Expand All @@ -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 {
Expand Down
7 changes: 2 additions & 5 deletions trogon-eventstore/src/operations/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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> {
Expand Down
4 changes: 2 additions & 2 deletions trogon-eventstore/src/types.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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>,
}
Expand Down
13 changes: 7 additions & 6 deletions trogon-eventstore/tests/api/operations.rs
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
use std::time::Duration;
use tracing::debug;
use trogon_eventstore::Credentials;
use trogon_eventstore::operations;
use trogon_eventstore::operations::StatsOptions;

Expand Down Expand Up @@ -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?;

Expand Down Expand Up @@ -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(())
}
Expand Down
40 changes: 16 additions & 24 deletions trogon-eventstore/tests/api/streams.rs
Original file line number Diff line number Diff line change
@@ -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};
Expand Down Expand Up @@ -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;

Expand All @@ -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>
});
Comment thread
coderabbitai[bot] marked this conversation as resolved.

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
);
}

Expand Down Expand Up @@ -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(())
}
Expand Down
20 changes: 5 additions & 15 deletions trogon-eventstore/tests/images.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -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
}
Expand Down Expand Up @@ -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![],
}
}
Expand Down
2 changes: 1 addition & 1 deletion trogon-eventstore/tests/integration.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading