Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: return bound endpoints after server started #113

Open
wants to merge 4 commits into
base: main
Choose a base branch
from
Open

fix: return bound endpoints after server started #113

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
eb1767a
fix: return bound endpoints after server started
kriswuollett May 12, 2023
41fd3d7
docs: fix wording of server methods
kriswuollett May 12, 2023
a46c7f0
Apply suggestions from code review
kriswuollett May 18, 2023
4648e46
fix: clarify copying of endpoints struct
kriswuollett May 18, 2023
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
123 changes: 85 additions & 38 deletions crates/server/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,8 @@ use std::{
sync::Arc,
time::Duration,
};
use tokio::task::JoinSet;
use tokio_util::sync::CancellationToken;
use tower_http::{
trace::{DefaultMakeSpan, DefaultOnResponse, TraceLayer},
LatencyUnit,
Expand All @@ -30,7 +32,7 @@ const DEFAULT_CHECKPOINT_INTERVAL: Duration = Duration::from_secs(5);

/// The server configuration.
pub struct Config {
operator_key: PrivateKey,
operator_key: Option<PrivateKey>,
addr: Option<SocketAddr>,
data_store: Option<Box<dyn datastore::DataStore>>,
content_dir: Option<PathBuf>,
Expand Down Expand Up @@ -58,7 +60,7 @@ impl Config {
/// Creates a new server configuration.
pub fn new(operator_key: PrivateKey) -> Self {
Self {
operator_key,
operator_key: Some(operator_key),
addr: None,
data_store: None,
content_dir: None,
Expand Down Expand Up @@ -120,60 +122,68 @@ impl Config {
/// Represents the warg registry server.
pub struct Server {
config: Config,
listener: Option<TcpListener>,
endpoints: Option<Endpoints>,
token: CancellationToken,
tasks: JoinSet<Result<()>>,
}

/// The bound endpoints for the warg registry server.
#[derive(Clone)]
pub struct Endpoints {
/// The address of the API endpoint.
pub api: SocketAddr,
kriswuollett marked this conversation as resolved.
Show resolved Hide resolved
}

impl Server {
/// Creates a new server with the given configuration.
pub fn new(config: Config) -> Self {
Self {
config,
listener: None,
token: CancellationToken::new(),
endpoints: None,
tasks: JoinSet::new(),
}
}

/// Binds the server to the configured address.
/// Starts the server and binds its endpoints to the configured addresses.
///
/// Returns the address the server bound to.
pub fn bind(&mut self) -> Result<SocketAddr> {
/// Returns the endpoints the server bound to.
pub async fn start(&mut self) -> Result<Endpoints> {
assert!(
self.endpoints.is_none(),
"cannot start server multiple times"
);
Comment on lines +152 to +155
Copy link
Member

@peterhuene peterhuene May 16, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So I'm personally not a big fan of taking &mut self for this as it leads to confusing internal state and needing to wrap things in Option just for this method to take their contents.

What if we renamed this start method to spawn and in addition to returning the endpoints, we return a type wrapping both the JoinSet representing the spawned server's tasks and the CancellationToken that can be used to stop the server. Thus, the type could have a stop method on it to stop the server and also implement Future that a caller can await on to join the tasks in the set?

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agree, I was just about to ask why required fields were becoming optional. I'd much rather have spawning produce a dedicated type that represents the info for a started service.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Was just trying to minimize the patch since existing code was &mut self and encountered issues like #112 when working on #111. Also wasn't sure why there was a struct called Server if it wasn't meant to be used after calling a start-like method.

Things became Option-like in config so self wasn't partially moved. If I had continued in same direction as the PR was going, I'd have flattened the Config's members into Server and not kept a config member in the Server struct.

See other comment about providing a CancellationToken to chain a shutdown procedure rather than explicitly telling dependencies to shut down.

Copy link
Member

@peterhuene peterhuene May 18, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The existing code for bind method is &mut self, but the former run method was self, both are now combined into your start method.

That said, I think perhaps we could better model our Server off of hyper's Server; theirs is already a bound and listening server upon construction thus there doesn't need to be a start (or spawn) method.

That is, you contruct/build a server, it internally binds based on its configuration; the binding can be inspected via a method; in the case of hype there is only a single bound address for the server, but our implementation might have multiple bound addresses as there might be multiple hyper servers internally (e.g. API, monitoring, etc).

Finally, it implements Future, so that you simply await it for it to run to graceful completion.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Cool, yes, I'd follow some patterns used on some existing servers already like the hyper used here already.


tracing::debug!(
"using server configuration: {config:?}",
config = self.config
);

let addr = self
.config
.addr
.to_owned()
.unwrap_or_else(|| DEFAULT_BIND_ADDRESS.parse().unwrap());

tracing::debug!("binding server to address `{addr}`");
tracing::debug!("binding api endpoint to address `{addr}`");
let listener = TcpListener::bind(addr)
.with_context(|| format!("failed to bind to address `{addr}`"))?;
.with_context(|| format!("failed to bind api endpoint to address `{addr}`"))?;

let addr = listener
.local_addr()
.context("failed to get local address for listen socket")?;

tracing::debug!("server bound to address `{addr}`");
self.config.addr = Some(addr);
self.listener = Some(listener);
Ok(addr)
}

/// Runs the server.
pub async fn run(mut self) -> Result<()> {
if self.listener.is_none() {
self.bind()?;
}

let listener = self.listener.unwrap();
.context("failed to get local address for api endpoint listen socket")?;
tracing::debug!("api endpoint bound to address `{addr}`");

tracing::debug!(
"using server configuration: {config:?}",
config = self.config
);
let endpoints = Endpoints { api: addr };
self.endpoints = Some(endpoints.clone());

let store = self
.config
.data_store
.take()
.unwrap_or_else(|| Box::<MemoryDataStore>::default());
let (core, handle) = CoreService::spawn(
self.config.operator_key,
self.config.operator_key.take().unwrap(),
store,
self.config
.checkpoint_interval
Expand All @@ -183,32 +193,69 @@ impl Server {

let server = axum::Server::from_tcp(listener)?.serve(
Self::create_router(
format!("http://{addr}", addr = self.config.addr.unwrap()),
self.config.content_dir,
format!("http://{addr}", addr = endpoints.api),
self.config.content_dir.take(),
core,
)?
.into_make_service(),
);

tracing::info!("listening on {addr}", addr = self.config.addr.unwrap());
tracing::info!("api endpoint listening on {addr}", addr = endpoints.api);

if let Some(shutdown) = self.config.shutdown {
tracing::debug!("server is running with a shutdown signal");
// Shut down core service when token cancelled.
let token = self.token.clone();
self.tasks.spawn(async move {
token.cancelled().await;
tracing::info!("waiting for core service to stop");
handle.stop().await;
Ok(())
});
Comment on lines +207 to +212
Copy link
Member

@peterhuene peterhuene May 16, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could this task be merged with the next task? That is to say, after the server future completes in the next task, we simply wait on the stop handle for the core service.

Given this design, I would expect a 1:1 between a task in the task set and a running Hyper server (API in this case, but soon a monitoring server, etc). Each would shutdown in their own way.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd think due to the nature of fanning out a cancelation signal chain, a cancellation token should be cloned for each owned depedency, somewhat similar to how Go Context works. Otherwise there virtually will be for-loop boilerplate being written to cancel "children" all separately as a server grows. That allows things to shut down in parallel easily if needed.

But yes, could keep things simple for now and merge the shutdown tasks instead of renaming chaining of them through the tokens themselves. However I'd either want to use a broadcast channel instead, or an additional token for draining in #111.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think we want the core service to be stopped in parallel to the hyper server shutting down as it might race.

If we call stop on the core service, it will break out of its loop and the task will complete; note that the CoreService instance itself won't be dropped as a result of this as there are references to it from the hyper server.

If this is done in parallel to the hyper server shutting down, then it's possible a request being drained attempts to enqueue work for the core service to do; if it so happens that the core service already stopped and the send limit for the message queue is reached, it'll block waiting for the core service to remove a message which will never occur since its task has ended.

We also don't want to close the message queue's sender as part of stopping the core service as this would result in a panic (even a graceful error back to the client here is not ideal) and a reset of the connection for any request attempting to enqueue work for the core service to do after it has stopped.

I believe the correct order of operations here is to shutdown the hyper server then join on the core service, like we had before.

Copy link
Member

@peterhuene peterhuene May 18, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Note that eventually the core service won't be part warg-server, but instead another backend service for it to talk to via IPC (i.e. a real distributed and reliable message queue) for processing new record entries; it's the limiting factor of why we can't spin up multiple warg-server instances right now.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I was merely suggesting making parallel shutdown possible, but did not point out a use case yet where to apply it now. I'm well aware about needing to keep dependencies going, like a core service, while a user of it drains, like the API server.

We share the same view of the shutdown order, but what is your thoughts on using a CancellationToken to signal some thing(s) to shut down, versus explicit function call on a dependency to shut down? But please note I'm not suggesting a change in this PR to inject a token into into CoreService itself. The handle.stop().await would remain. This helps a bit when/if drain needs to be handled before an actual shutdown.


// Shut down server when token cancelled.
let token: CancellationToken = self.token.clone();
self.tasks.spawn(async move {
tracing::info!("waiting for api endpoint to stop");
server
.with_graceful_shutdown(async move { shutdown.await })
.with_graceful_shutdown(async move { token.cancelled().await })
.await?;
Ok(())
});

// Cancel token if shutdown signal received.
if let Some(shutdown) = self.config.shutdown.take() {
tracing::debug!("server is running with a shutdown signal");
let token = self.token.clone();
tokio::spawn(async move {
tracing::info!("waiting for shutdown signal");
shutdown.await;
tracing::info!("shutting down server");
token.cancel();
});
} else {
tracing::debug!("server is running without a shutdown signal");
server.await?;
}

tracing::info!("waiting for core service to stop");
handle.stop().await;
Ok(endpoints)
}

/// Waits on a started server to shutdown.
pub async fn join(&mut self) -> Result<()> {
while (self.tasks.join_next().await).is_some() {}
tracing::info!("server shutdown complete");
Ok(())
}

/// Starts the server and waits for completion.
pub async fn run(&mut self) -> Result<()> {
self.start().await?;
self.join().await?;
Ok(())
}

pub fn stop(&mut self) {
self.token.cancel();
}

fn create_router(
base_url: String,
content_dir: Option<PathBuf>,
Expand Down
1 change: 0 additions & 1 deletion tests/server.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@ use std::{fs, str::FromStr};
use warg_client::{api, Config, FileSystemClient, StorageLockResult};
use warg_crypto::{signing::PrivateKey, Encode, Signable};
use wit_component::DecodedWasm;

mod support;

#[cfg(feature = "postgres")]
Expand Down
6 changes: 3 additions & 3 deletions tests/support/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,10 +82,10 @@ pub async fn spawn_server(
}

let mut server = Server::new(config);
let addr = server.bind()?;
let endpoints = server.start().await?;

let task = tokio::spawn(async move {
server.run().await.unwrap();
server.join().await.unwrap();
});

let instance = ServerInstance {
Expand All @@ -94,7 +94,7 @@ pub async fn spawn_server(
};

let config = warg_client::Config {
default_url: Some(format!("http://{addr}")),
default_url: Some(format!("http://{addr}", addr = endpoints.api)),
registries_dir: Some(root.join("registries")),
content_dir: Some(root.join("content")),
};
Expand Down