1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
//! The [`MockNexus`] handle: start, stop, store access and the request log.
use std::io;
use std::net::{SocketAddr, TcpListener, TcpStream};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use crate::http;
use crate::scenario::Scenario;
use crate::store::{lock, Outcome, ReqLog, Shared};
/// Handle to a running mock server.
/// Dropping it stops the accept loop and releases the port.
/// In-flight connection threads finish on their own.
#[derive(Debug)]
pub struct MockNexus {
shared: Arc<Shared>,
addr: SocketAddr,
stop: Arc<AtomicBool>,
}
impl MockNexus {
/// Bind a server on a random free port on 127.0.0.1.
pub fn start(scenario: Scenario) -> std::io::Result<Self> {
Self::start_on(scenario, SocketAddr::from(([127, 0, 0, 1], 0)))
}
/// Start a hosted instance from a composable [`Spec`](crate::Spec) instead
/// of a named [`Scenario`] preset.
///
/// The spec layers stack: `Spec::atomic().with_auth(..).with_flaky(3)` is
/// an auth-gated server that also answers the first three requests per
/// path with 503, which no single preset expresses.
/// Content arms (the service documents, the seeded litter) are preset
/// territory and stay on the enum: a spec-driven instance serves the plain
/// atomic service surface.
/// A composed instance reports `scenario_name() == "spec"`.
pub fn start_spec(spec: crate::spec::Spec) -> std::io::Result<Self> {
Self::start_spec_on(spec, SocketAddr::from(([127, 0, 0, 1], 0)))
}
/// Like [`MockNexus::start_spec`] but bound to `addr`.
pub fn start_spec_on(spec: crate::spec::Spec, addr: SocketAddr) -> std::io::Result<Self> {
Self::start_generic(Some(spec), None, addr)
}
/// Like [`MockNexus::start`] but bound to `addr` (the binary's `--port`).
pub fn start_on(scenario: Scenario, addr: SocketAddr) -> std::io::Result<Self> {
Self::start_generic(None, Some(scenario), addr)
}
/// The one hosted constructor: a spec, a preset scenario, or both.
/// `None, None` is the group base (an atomic spec over an empty store).
fn start_generic(
spec: Option<crate::spec::Spec>,
scenario: Option<Scenario>,
addr: SocketAddr,
) -> std::io::Result<Self> {
let listener = TcpListener::bind(addr)?;
let bound = listener.local_addr()?;
let spec_driven = spec.is_some();
let scenario = scenario.unwrap_or(Scenario::Atomic);
let spec = spec.unwrap_or_else(|| crate::spec::Spec::from(&scenario));
let shared = Arc::new(Shared::hosted_explicit(
spec,
scenario,
spec_driven,
format!("http://{bound}/"),
));
let stop = Arc::new(AtomicBool::new(false));
spawn_acceptor(listener, Arc::clone(&shared), Arc::clone(&stop));
Ok(Self {
shared,
addr: bound,
stop,
})
}
/// The scenario preset this instance runs, or `"spec"` for a composed one.
pub fn scenario_name(&self) -> &str {
if self.shared.spec_driven {
"spec"
} else {
self.shared.scenario.as_str()
}
}
/// Start a group repository over `members` (two or more running instances), listening on its own port on 127.0.0.1.
///
/// Reads (`GET`/`HEAD`) are forwarded to the members' real HTTP endpoints in member order, and the first member answering `2xx` is relayed to the client.
/// Any other member answer (a `404`, a `5xx`, a broken connection) is skipped: the walk never retries a member and never surfaces a member's failure to the client.
/// When no member answers `2xx`, the group answers `404` with the same `not found` body as a single-instance miss (real Nexus's `notFound()` sends no body, the mock stays consistent with its own single-repo shape).
/// Every other method is refused with `405`, `Allow: GET,HEAD` and an empty body: a group is a read-only aggregation, and publication targets hosted members.
/// Member scenario gates apply to forwarded requests, because the members serve them for real: a flaky member's `503` costs the group nothing once a later member holds the object.
///
/// Limitations, chosen to keep the mock std-only and faithful to the real `GroupHandler` dispatch: members are flat (no nested groups), every member is assumed online, the format-specific escape hatches are not modeled, and the search API is not dispatched (a separate subsystem in real Nexus, so a group URL answers `404` for it).
///
/// # Errors
///
/// Returns an [`io::Error`] of kind [`io::ErrorKind::InvalidInput`] for fewer than two members, and the bind error when the listener cannot open.
pub fn start_group(members: &[&MockNexus]) -> std::io::Result<Self> {
if members.len() < 2 {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
"a group repository needs at least two members",
));
}
let listener = TcpListener::bind(SocketAddr::from(([127, 0, 0, 1], 0)))?;
let bound = listener.local_addr()?;
let shared = Arc::new(Shared::group(members));
let stop = Arc::new(AtomicBool::new(false));
spawn_acceptor(listener, Arc::clone(&shared), Arc::clone(&stop));
Ok(Self {
shared,
addr: bound,
stop,
})
}
/// The address the server is bound to.
pub fn addr(&self) -> SocketAddr {
self.addr
}
/// Base URL for clients: `http://<addr>/` with a trailing slash.
pub fn base_url(&self) -> String {
format!("http://{}/", self.addr)
}
/// Stored bytes for `path` (no leading `/`, e.g. `1.14.0/version.json`).
pub fn store_get(&self, path: &str) -> Option<Vec<u8>> {
lock(&self.shared.store).get(path).cloned()
}
/// Seed the store directly, bypassing all scenario behavior.
pub fn insert(&self, path: &str, bytes: &[u8]) {
lock(&self.shared.store).insert(path.to_owned(), bytes.to_vec());
}
/// Every request so far, in arrival order (including failures).
pub fn requests(&self) -> Vec<ReqLog> {
lock(&self.shared.log).clone()
}
/// Number of PUT requests for `path` that got a 2xx response.
pub fn put_count(&self, path: &str) -> usize {
self.requests()
.iter()
.filter(|req| {
req.method == "PUT"
&& req.path == path
&& matches!(req.outcome, Outcome::Status(200..=299))
})
.count()
}
/// Start serving synthesized (drifted) documents for `*/version.json` GETs.
pub fn enable_drift(&self) {
self.shared.drift.store(true, Ordering::Relaxed);
}
/// Return to store-backed document responses.
pub fn disable_drift(&self) {
self.shared.drift.store(false, Ordering::Relaxed);
}
}
/// One accept-error kind is reusable: aborted or interrupted waits, plus the
/// Windows surface of the same race (`WSAECONNRESET` for a dropped pending peer).
fn is_transient_accept(e: &io::Error) -> bool {
matches!(
e.kind(),
io::ErrorKind::ConnectionAborted
| io::ErrorKind::ConnectionReset
| io::ErrorKind::Interrupted
)
}
/// Accept connections until `stop` is set.
/// One thread per connection.
fn spawn_acceptor(listener: TcpListener, shared: Arc<Shared>, stop: Arc<AtomicBool>) {
std::thread::spawn(move || loop {
match listener.accept() {
Ok((conn, _)) => {
if stop.load(Ordering::Relaxed) {
break;
}
let conn_shared = Arc::clone(&shared);
std::thread::spawn(move || http::serve(&conn_shared, conn));
}
// A burst of short-lived connections lets some abort between the
// handshake and this accept (ConnectionReset covers the Windows
// surface of the same race): the name is reusable, take the next one.
Err(e) if is_transient_accept(&e) => {
eprintln!("mock: accept rode out a transient error: {e}");
}
Err(e) => {
eprintln!("mock: acceptor stopping on a fatal accept error: {e}");
break;
}
}
});
}
impl Drop for MockNexus {
fn drop(&mut self) {
self.stop.store(true, Ordering::Relaxed);
// Unblock the blocking accept() so the listener thread exits and the port is released.
let _ = TcpStream::connect(self.addr);
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn transient_accept_classification() {
let transient = |kind| is_transient_accept(&io::Error::new(kind, "probe"));
assert!(transient(io::ErrorKind::ConnectionAborted));
assert!(transient(io::ErrorKind::ConnectionReset));
assert!(transient(io::ErrorKind::Interrupted));
assert!(!transient(io::ErrorKind::PermissionDenied));
assert!(!transient(io::ErrorKind::AddrInUse));
}
}