Skip to content

Commit dcb755b

Browse files
jaredLundeclaude
andcommitted
implement Phase 4: S3-compatible surface
Adds an S3-compatible wire protocol via the s3s crate, mounted as fallback_service alongside the existing /v1/* REST routes. SigV4 maps onto the existing HMAC-derived bucket token scheme so AWS SDKs work with (accessKeyId: bucket, secretAccessKey: deriveToken(root, bucket)). Multipart uploads are exposed only on the S3 surface; storage gains init/write_part/complete/abort/list with crash-safe rename semantics and GC for orphaned upload dirs. Adds deriveS3Credentials to the TS SDK and refreshes DESIGN.md plus a new ARCHITECTURE.md. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1 parent 22a321b commit dcb755b

31 files changed

Lines changed: 4296 additions & 212 deletions

‎ARCHITECTURE.md‎

Lines changed: 355 additions & 0 deletions
Large diffs are not rendered by default.

‎Cargo.lock‎

Lines changed: 1006 additions & 54 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎Cargo.toml‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ license = "MIT"
88

99
[workspace.dependencies]
1010
anyhow = "1"
11+
async-trait = "0.1"
1112
axum = { version = "0.8", features = ["macros"] }
1213
bytes = "1"
1314
chrono = { version = "0.4", features = ["serde"] }
@@ -23,6 +24,7 @@ opentelemetry-otlp = { version = "0.31", features = ["grpc-tonic"] }
2324
opentelemetry_sdk = { version = "0.31", features = ["rt-tokio"] }
2425
prometheus = "0.14"
2526
reqwest = { version = "0.13", default-features = false, features = ["json", "rustls", "stream"] }
27+
s3s = "0.13"
2628
secrecy = "0.8"
2729
serde = { version = "1", features = ["derive"] }
2830
serde_json = "1"

‎DESIGN.md‎

Lines changed: 44 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -24,10 +24,27 @@ Clean JSON API, Beyond bearer-token auth, no XML, no Sig V4. This is what the Ty
2424

2525
### S3-compatible
2626

27-
Full S3 wire protocol via [`s3s`](https://github.com/s3s-project/s3s) — a hyper middleware that handles S3 routing, request parsing, multipart lifecycle, and error serialization. We implement one `S3` trait backed by the filesystem.
27+
Full S3 wire protocol via [`s3s`](https://github.com/Nugine/s3s) — a tower service that handles S3 routing, request parsing, multipart lifecycle, and error serialization. We implement one `S3` trait backed by the same filesystem.
2828

2929
S3 "buckets" map directly to Beyond buckets on the filesystem. `CreateBucket` is a `mkdir` + xattr init. `ListBuckets` reads top-level directories.
3030

31+
**Auth — SigV4 mapped to our HMAC scheme.** AWS SDKs always sign with `(access_key_id, secret_access_key)`. We answer s3s's "what's the secret for this access_key?" question by reusing the existing token derivation:
32+
33+
| access_key_id | secret_access_key |
34+
| ------------- | ------------------------------------------------- |
35+
| `"root"` | `OBJECTS_ROOT_TOKEN` |
36+
| `<bucket>` | `HMAC-SHA256(OBJECTS_ROOT_TOKEN, <bucket>)` (hex) |
37+
38+
So an S3 client using `accessKeyId: "images"` and `secretAccessKey: <hex>` is byte-for-byte equivalent to a REST client sending `Authorization: Bearer <hex>`. Authorization (which buckets a key may touch) is enforced by `S3Access`: `access_key == bucket || access_key == "root"`. The TS SDK exports `deriveS3Credentials(rootToken, bucket)` for symmetry with `deriveToken`.
39+
40+
**Path-style only.** `https://{endpoint}/{bucket}/{key…}`. Virtual-hosted-style is not supported (it would require a wildcard cert per project, and AWS clients support `forcePathStyle: true`).
41+
42+
**Anonymous reads.** Public objects can be GET/HEAD'd without credentials, mirroring the REST surface. Anonymous writes are denied at `S3Access`.
43+
44+
**What we support:** `ListBuckets`, `CreateBucket`, `DeleteBucket`, `HeadBucket`; `PutObject`, `GetObject`, `HeadObject`, `DeleteObject`, `CopyObject`, `ListObjectsV2`; the full multipart five (`CreateMultipartUpload`, `UploadPart`, `CompleteMultipartUpload`, `AbortMultipartUpload`, `ListMultipartUploads`, `ListParts`). Within those: `Range`, `If-Match`, `If-None-Match`, `If-Modified-Since`, `If-Unmodified-Since`, `Content-Type`, `x-amz-meta-*`, and `x-amz-acl: public-read | private` (mapped to the same `user.access` xattr the REST surface writes).
45+
46+
**What returns `NotImplemented`:** versioning, lifecycle, replication, CORS-config-via-API, website config, object lock, retention, tagging, presigned URLs, bucket policies, inventory, intelligent-tiering, analytics, encryption configuration, full ACL grants beyond the canned `private` / `public-read`.
47+
3148
---
3249

3350
## Storage
@@ -128,7 +145,24 @@ Supported via standard HTTP conditional headers on PUT:
128145

129146
### Multipart uploads (S3-compat only)
130147

131-
Multipart is not exposed in the native API — streaming handles large uploads directly. The S3-compatible layer supports it for clients that require it. Parts written to `.multipart/{upload_id}/{part_number}`; assembled on `CompleteMultipartUpload` via sequential concatenation → `fsync` → `rename()` → fjall INSERT.
148+
Multipart is not exposed in the native API — streaming handles large uploads directly. The S3-compatible layer supports it for clients that require it.
149+
150+
**On-disk layout:**
151+
152+
```
153+
{data_dir}/.multipart/{upload_id}/.meta.json {bucket, key, content_type, access, user_metadata, init_time_secs}
154+
{data_dir}/.multipart/{upload_id}/{part_n} raw part bytes; quoted-MD5 etag in `user.etag` xattr
155+
```
156+
157+
`CompleteMultipartUpload`: read the meta sidecar → concatenate parts in caller-supplied order to `.tmp/{uuid}` while computing `MD5({part_md5}|{part_md5}|…)` → fsync → set xattrs from meta → atomic `rename()` to `{bucket}/{key}` → fjall INSERT → `rm -rf .multipart/{upload_id}`. Final etag is `"{md5_of_part_md5s}-{N}"` (AWS convention). `AbortMultipartUpload` is `rm -rf .multipart/{upload_id}` (idempotent).
158+
159+
**Crash recovery:**
160+
161+
- Crash mid-`UploadPart` → tmp file in `.tmp/`, GC'd by `gc_temp_files`.
162+
- Crash after `complete` rename, before `.multipart/{id}` cleanup → orphan dir, GC'd by `gc_multipart_uploads`.
163+
- Crash before rename → object never appears; abort or GC sweeps it.
164+
165+
The `.multipart/` directory is dot-prefixed so `list_buckets` skips it automatically.
132166

133167
---
134168

@@ -146,6 +180,14 @@ const objects = createObjectsClient();
146180
// derive a scoped token to hand off to a specific service (HMAC-SHA256, async)
147181
const imagesToken = await deriveToken(process.env.OBJECTS_ROOT_TOKEN, "images");
148182
const images = createObjectsClient({ bucket: "images", token: imagesToken });
183+
184+
// the same secret, formatted for any AWS S3 SDK
185+
import { deriveS3Credentials } from "@beyond.dev/objects";
186+
const creds = await deriveS3Credentials(
187+
process.env.OBJECTS_ROOT_TOKEN,
188+
"images",
189+
);
190+
// { accessKeyId: "images", secretAccessKey: <hex> }
149191
```
150192

151193
**Per-object visibility**, set at write time, inherited from bucket default if absent:

‎crates/server/Cargo.toml‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ beyond-objects-index = { path = "../index" }
1717
beyond-objects-storage = { path = "../storage" }
1818

1919
anyhow = { workspace = true }
20+
async-trait = { workspace = true }
2021
axum = { workspace = true }
2122
bytes = { workspace = true }
2223
chrono = { workspace = true }
@@ -29,6 +30,7 @@ opentelemetry = { workspace = true }
2930
opentelemetry-otlp = { workspace = true }
3031
opentelemetry_sdk = { workspace = true }
3132
prometheus = { workspace = true }
33+
s3s = { workspace = true }
3234
secrecy = { workspace = true }
3335
serde = { workspace = true }
3436
serde_json = { workspace = true }
@@ -49,9 +51,18 @@ uuid = { workspace = true }
4951
tikv-jemallocator = { workspace = true }
5052

5153
[dev-dependencies]
54+
aws-config = { version = "1", default-features = false, features = ["behavior-version-latest", "rustls", "rt-tokio"] }
55+
aws-sdk-s3 = { version = "1", default-features = false, features = ["behavior-version-latest", "rustls", "rt-tokio"] }
56+
aws-smithy-async = { version = "1", features = ["rt-tokio"] }
57+
aws-smithy-runtime = { version = "1", features = ["client", "connector-hyper-0-14-x", "rt-tokio"] }
58+
aws-smithy-types = { version = "1", features = ["http-body-1-x"] }
5259
reqwest = { workspace = true }
5360
tempfile = { workspace = true }
5461

5562
[[test]]
5663
name = "rest"
5764
path = "tests/rest/main.rs"
65+
66+
[[test]]
67+
name = "s3"
68+
path = "tests/s3/main.rs"

‎crates/server/src/error.rs‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -71,6 +71,12 @@ pub enum ApiError {
7171
#[error("range not satisfiable")]
7272
RangeNotSatisfiable,
7373

74+
#[error("multipart upload not found: {0}")]
75+
UploadNotFound(String),
76+
77+
#[error("invalid part: {0}")]
78+
InvalidPart(String),
79+
7480
#[error("internal error: {0}")]
7581
Internal(#[from] anyhow::Error),
7682
}
@@ -91,6 +97,8 @@ impl From<StorageError> for ApiError {
9197
StorageError::EtagMismatch => Self::EtagMismatch,
9298
StorageError::InvalidKey(k) => Self::InvalidKey(k),
9399
StorageError::InvalidValue(v) => Self::BadRequest(v),
100+
StorageError::UploadNotFound(id) => Self::UploadNotFound(id),
101+
StorageError::InvalidPart(msg) => Self::InvalidPart(msg),
94102
StorageError::Xattr(msg) => Self::Internal(anyhow::anyhow!("xattr failure: {msg}")),
95103
StorageError::Io(e) => Self::Internal(anyhow::anyhow!("io: {e}")),
96104
}
@@ -134,6 +142,10 @@ impl IntoResponse for ApiError {
134142
"range_not_satisfiable",
135143
self.to_string(),
136144
),
145+
ApiError::UploadNotFound(_) => {
146+
(StatusCode::NOT_FOUND, "upload_not_found", self.to_string())
147+
}
148+
ApiError::InvalidPart(_) => (StatusCode::BAD_REQUEST, "invalid_part", self.to_string()),
137149
ApiError::Internal(_) => (
138150
StatusCode::INTERNAL_SERVER_ERROR,
139151
"internal_error",

‎crates/server/src/lib.rs‎

Lines changed: 100 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ pub mod error;
44
pub mod metrics;
55
pub mod middleware;
66
pub mod routes;
7+
pub mod s3;
78
pub mod telemetry;
89
pub mod test_support;
910

@@ -67,6 +68,79 @@ impl AppState {
6768
/// now — kept on AppState so handlers don't need to know whether events are
6869
/// wired up.
6970
pub fn publish(&self, _base_url: &str, _bucket: &str, _key: &str) {}
71+
72+
/// One page of a prefix-paginated listing. Used by both the native REST
73+
/// handler and the S3 `ListObjectsV2` handler so they share a single
74+
/// scan plus buffered-head implementation. Index entries whose backing
75+
/// file disappeared (typically after a crash) are skipped silently;
76+
/// startup reconcile drops them on the next boot.
77+
pub async fn list_page(
78+
&self,
79+
bucket: &str,
80+
prefix: &str,
81+
cursor: Option<&str>,
82+
limit: usize,
83+
) -> Result<ListPage, error::ApiError> {
84+
use beyond_objects_storage::StorageError;
85+
use futures::TryStreamExt;
86+
use futures::stream::StreamExt;
87+
88+
let index = self.index.clone();
89+
let bucket_owned = bucket.to_owned();
90+
let prefix_owned = prefix.to_owned();
91+
let cursor_owned = cursor.map(str::to_owned);
92+
let keys = tokio::task::spawn_blocking(move || {
93+
index.scan(&bucket_owned, &prefix_owned, cursor_owned.as_deref(), limit)
94+
})
95+
.await
96+
.map_err(|e| error::ApiError::Internal(anyhow::anyhow!("index scan join: {e}")))??;
97+
98+
let next_cursor = if keys.len() == limit {
99+
keys.last().cloned()
100+
} else {
101+
None
102+
};
103+
104+
let bucket_owned = bucket.to_owned();
105+
let items: Vec<ListItem> = futures::stream::iter(keys.into_iter().map(|k| {
106+
let storage = self.storage.clone();
107+
let b = bucket_owned.clone();
108+
async move {
109+
match storage.head_object(&b, &k).await {
110+
Ok(info) => Ok(Some(ListItem { key: k, info })),
111+
Err(StorageError::NotFound { .. }) => {
112+
tracing::debug!(
113+
bucket = %b,
114+
key = %k,
115+
"skipping index entry without backing file"
116+
);
117+
Ok::<Option<ListItem>, error::ApiError>(None)
118+
}
119+
Err(e) => Err(error::ApiError::from(e)),
120+
}
121+
}
122+
}))
123+
.buffered(64)
124+
.try_collect::<Vec<_>>()
125+
.await?
126+
.into_iter()
127+
.flatten()
128+
.collect();
129+
130+
Ok(ListPage { items, next_cursor })
131+
}
132+
}
133+
134+
/// One key+metadata pair returned by `AppState::list_page`. Surface layers
135+
/// (REST, S3) wrap this into their own response shapes.
136+
pub struct ListItem {
137+
pub key: String,
138+
pub info: beyond_objects_storage::ObjectInfo,
139+
}
140+
141+
pub struct ListPage {
142+
pub items: Vec<ListItem>,
143+
pub next_cursor: Option<String>,
70144
}
71145

72146
#[derive(Clone)]
@@ -83,8 +157,14 @@ impl MakeRequestId for MakeRequestUuid {
83157
/// sit outside the auth boundary; per-resource auth is applied inside
84158
/// `routes::router`. The `/metrics` endpoint lives on a separate internal-only
85159
/// listener built via `build_metrics_router`.
160+
///
161+
/// The S3-compatible surface is mounted as `fallback_service` so that explicit
162+
/// `/v1/*`, `/healthz`, and `/v1/openapi.json` routes always win. Any
163+
/// unmatched URL (e.g. `GET /` for `ListBuckets`, `PUT /{bucket}/{key}` for
164+
/// `PutObject`) is handed to s3s.
86165
pub fn build_router(state: AppState) -> Router {
87166
let openapi = routes::ApiDoc::openapi();
167+
let s3_fallback = s3::service(state.clone());
88168

89169
routes::router(state.clone())
90170
.route(
@@ -96,8 +176,13 @@ pub fn build_router(state: AppState) -> Router {
96176
)
97177
.route("/healthz", get(routes::healthz::handler))
98178
.with_state(state.clone())
179+
.fallback_service(
180+
ServiceBuilder::new()
181+
.layer(DefaultBodyLimit::disable())
182+
.service(s3_fallback),
183+
)
99184
.route_layer(from_fn_with_state(state, record_metrics))
100-
.layer(DefaultBodyLimit::max(64 * 1024))
185+
.route_layer(DefaultBodyLimit::max(64 * 1024))
101186
.layer(
102187
ServiceBuilder::new()
103188
.layer(SetRequestIdLayer::x_request_id(MakeRequestUuid))
@@ -197,7 +282,7 @@ pub async fn serve(config: Config) -> Result<()> {
197282
config: Arc::new(config),
198283
storage,
199284
index,
200-
metrics: Arc::new(metrics::Metrics::new()),
285+
metrics: Arc::new(metrics::Metrics::try_new()?),
201286
};
202287

203288
let app = build_router(state.clone());
@@ -221,17 +306,23 @@ async fn shutdown_signal() {
221306
use tokio::signal;
222307

223308
let ctrl_c = async {
224-
signal::ctrl_c()
225-
.await
226-
.expect("failed to install Ctrl+C handler");
309+
if let Err(e) = signal::ctrl_c().await {
310+
tracing::warn!(err = %e, "Ctrl+C handler unavailable; relying on SIGTERM");
311+
std::future::pending::<()>().await;
312+
}
227313
};
228314

229315
#[cfg(unix)]
230316
let terminate = async {
231-
signal::unix::signal(signal::unix::SignalKind::terminate())
232-
.expect("failed to install SIGTERM handler")
233-
.recv()
234-
.await;
317+
match signal::unix::signal(signal::unix::SignalKind::terminate()) {
318+
Ok(mut sig) => {
319+
sig.recv().await;
320+
}
321+
Err(e) => {
322+
tracing::warn!(err = %e, "SIGTERM handler unavailable; relying on Ctrl+C");
323+
std::future::pending::<()>().await;
324+
}
325+
}
235326
};
236327

237328
#[cfg(not(unix))]

‎crates/server/src/metrics.rs‎

Lines changed: 7 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -7,14 +7,13 @@ pub struct Metrics {
77
}
88

99
impl Metrics {
10-
pub fn new() -> Self {
10+
pub fn try_new() -> Result<Self, prometheus::Error> {
1111
let registry = Registry::new();
1212

1313
let http_requests_total = CounterVec::new(
1414
Opts::new("http_requests_total", "Total HTTP requests"),
1515
&["method", "path", "status"],
16-
)
17-
.expect("valid metric");
16+
)?;
1817

1918
let http_request_duration_seconds = HistogramVec::new(
2019
histogram_opts!(
@@ -23,21 +22,16 @@ impl Metrics {
2322
vec![0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5]
2423
),
2524
&["method", "path"],
26-
)
27-
.expect("valid metric");
25+
)?;
2826

29-
registry
30-
.register(Box::new(http_requests_total.clone()))
31-
.expect("unique metric");
32-
registry
33-
.register(Box::new(http_request_duration_seconds.clone()))
34-
.expect("unique metric");
27+
registry.register(Box::new(http_requests_total.clone()))?;
28+
registry.register(Box::new(http_request_duration_seconds.clone()))?;
3529

36-
Self {
30+
Ok(Self {
3731
http_requests_total,
3832
http_request_duration_seconds,
3933
registry,
40-
}
34+
})
4135
}
4236

4337
pub fn render(&self) -> Result<String, prometheus::Error> {

0 commit comments

Comments
 (0)