Release ObjectStore 0.0.3

README.md

@@ -8,7 +8,7 @@ Source: [GitHub](https://github.com/LunarSkyOSS/ObjectStore) · [Gitea mirror](h
 
 **Development software. Do not use it for production data.** It is provided as is, without warranty, under the [MIT License](LICENSE).
 
-[![Stage: development](assets/badge-stage.svg)](#limits-and-safety) [![License: MIT](assets/badge-license.svg)](LICENSE) [![Runtime: Java 21](assets/badge-java.svg)](Dockerfile) [![S3 API: partial support](assets/badge-api.svg)](#s3-api-support-checklist)
+[![Stage: development](assets/badge-stage.svg)](#limits-and-safety) [![License: MIT](assets/badge-license.svg)](LICENSE) [![Runtime: Java 21](assets/badge-java.svg)](Dockerfile) [![S3 API: partial support](assets/badge-api.svg)](#s3-api-support-checklist) [![CodeFactor](https://www.codefactor.io/repository/github/lunarskyoss/objectstore/badge)](https://www.codefactor.io/repository/github/lunarskyoss/objectstore)
 
 ## Contents
 

scripts/test-cluster-http.py

@@ -2,11 +2,10 @@
 import datetime
 import hashlib
 import hmac
+import http.client
 import pathlib
 import sys
-import urllib.error
 import urllib.parse
-import urllib.request
 
 
 values = dict(line.strip().split("=", 1) for line in pathlib.Path(sys.argv[1]).read_text().splitlines()
@@ -14,7 +13,9 @@ values = dict(line.strip().split("=", 1) for line in pathlib.Path(sys.argv[1]).r
 access = values["S3_ACCESS_KEY"]
 secret = values["S3_SECRET_KEY"]
 bucket = values.get("S3_BUCKET", "objects")
-port = values.get("CLUSTER_HOST_PORT", "9001")
+port = int(values.get("CLUSTER_HOST_PORT", "9001"))
+if not 1 <= port <= 65535:
+    raise ValueError("CLUSTER_HOST_PORT must be between 1 and 65535")
 host = f"127.0.0.1:{port}"
 
 
@@ -40,14 +41,13 @@ def request(method, path, body=b"", extra=None):
     signature = hmac.new(key, to_sign.encode(), hashlib.sha256).hexdigest()
     headers["authorization"] = (f"AWS4-HMAC-SHA256 Credential={access}/{scope},"
                                 f"SignedHeaders={signed_names},Signature={signature}")
-    url = f"http://{host}{path}"
+    connection = http.client.HTTPConnection("127.0.0.1", port, timeout=30)
-    outgoing = urllib.request.Request(url, data=body if method == "PUT" else None,
-                                      method=method, headers=headers)
     try:
-        with urllib.request.urlopen(outgoing, timeout=30) as response:
+        connection.request(method, path, body=body if method == "PUT" else None, headers=headers)
-            return response.status, response.read(), response.headers
+        response = connection.getresponse()
-    except urllib.error.HTTPError as error:
+        return response.status, response.read(), response.headers
-        return error.code, error.read(), error.headers
+    finally:
+        connection.close()
 
 
 if len(sys.argv) > 2 and sys.argv[2] == "survivor":

src/cloud/lunarsky/store/ClusterNode.java

@@ -78,40 +78,14 @@ public final class ClusterNode implements AutoCloseable {
                 respond(exchange, 200, "ok");
                 return;
             }
-            byte[] supplied = exchange.getRequestHeaders().getFirst("X-Cluster-Token") == null
+            if (!authorized(exchange)) return;
-                ? new byte[0] : exchange.getRequestHeaders().getFirst("X-Cluster-Token")
-                    .getBytes(java.nio.charset.StandardCharsets.UTF_8);
-            if (!MessageDigest.isEqual(token, supplied)) {
-                respond(exchange, 403, "Forbidden");
-                return;
-            }
             if (path.equals("/identity") && exchange.getRequestMethod().equals("GET")) {
                 respond(exchange, 200, identity.nodeId() + " " + identity.hostId());
                 return;
             }
-            if (!identity.nodeId().toString().equals(exchange.getRequestHeaders().getFirst("X-Cluster-Expected-Node"))) {
+            if (!expectedNodeAndRepairAuthorized(exchange)) return;
-                respond(exchange, 409, "Wrong storage node");
+            String id = segmentId(exchange, path);
-                return;
+            if (id == null) return;
-            }
-            if (exchange.getRequestMethod().equals("PUT") &&
-                "true".equals(exchange.getRequestHeaders().getFirst("X-Cluster-Repair"))) {
-                String suppliedRepair = exchange.getRequestHeaders().getFirst("X-Cluster-Repair-Token");
-                byte[] suppliedBytes = suppliedRepair == null ? new byte[0]
-                    : suppliedRepair.getBytes(java.nio.charset.StandardCharsets.UTF_8);
-                if (!MessageDigest.isEqual(repairToken, suppliedBytes)) {
-                    respond(exchange, 403, "Repair authority required");
-                    return;
-                }
-            }
-            if (!path.matches("/segments/[0-9a-f-]{36}")) {
-                respond(exchange, 404, "Not found");
-                return;
-            }
-            String id = path.substring("/segments/".length());
-            if (!UUID.fromString(id).toString().equals(id)) {
-                respond(exchange, 400, "Invalid segment ID");
-                return;
-            }
             switch (exchange.getRequestMethod()) {
                 case "PUT" -> put(exchange, segmentPath(id, true));
                 case "GET" -> get(exchange, segmentPath(id, false));
@@ -127,6 +101,48 @@ public final class ClusterNode implements AutoCloseable {
         }
     }
 
+    private boolean authorized(HttpExchange exchange) throws IOException {
+        byte[] supplied = exchange.getRequestHeaders().getFirst("X-Cluster-Token") == null
+            ? new byte[0] : exchange.getRequestHeaders().getFirst("X-Cluster-Token")
+                .getBytes(java.nio.charset.StandardCharsets.UTF_8);
+        if (!MessageDigest.isEqual(token, supplied)) {
+            respond(exchange, 403, "Forbidden");
+            return false;
+        }
+        return true;
+    }
+
+    private boolean expectedNodeAndRepairAuthorized(HttpExchange exchange) throws IOException {
+        if (!identity.nodeId().toString().equals(exchange.getRequestHeaders().getFirst("X-Cluster-Expected-Node"))) {
+            respond(exchange, 409, "Wrong storage node");
+            return false;
+        }
+        if (exchange.getRequestMethod().equals("PUT") &&
+            "true".equals(exchange.getRequestHeaders().getFirst("X-Cluster-Repair"))) {
+            String suppliedRepair = exchange.getRequestHeaders().getFirst("X-Cluster-Repair-Token");
+            byte[] suppliedBytes = suppliedRepair == null ? new byte[0]
+                : suppliedRepair.getBytes(java.nio.charset.StandardCharsets.UTF_8);
+            if (!MessageDigest.isEqual(repairToken, suppliedBytes)) {
+                respond(exchange, 403, "Repair authority required");
+                return false;
+            }
+        }
+        return true;
+    }
+
+    private static String segmentId(HttpExchange exchange, String path) throws IOException {
+        if (!path.matches("/segments/[0-9a-f-]{36}")) {
+            respond(exchange, 404, "Not found");
+            return null;
+        }
+        String id = path.substring("/segments/".length());
+        if (!UUID.fromString(id).toString().equals(id)) {
+            respond(exchange, 400, "Invalid segment ID");
+            return null;
+        }
+        return id;
+    }
+
     private synchronized Path segmentPath(String id, boolean createShard) throws IOException {
         Path shard = segments.resolve(id.substring(0, 2));
         if (createShard && !Files.isDirectory(shard)) {

src/cloud/lunarsky/store/ClusterStore.java

@@ -52,6 +52,23 @@ final class ClusterStore implements ObjectStorage {
 
     @Override public Metadata put(String bucket, String key, InputStream input, long length, String expectedHash,
                                   String checksum, boolean createOnly, String contentType) throws IOException {
+        validatePut(bucket, length, contentType);
+        MessageDigest md5 = digest("MD5");
+        Path staged = Files.createTempFile("objectstore-cluster-", ".pending");
+        List<Segment> segments;
+        byte[] fullHash;
+        try {
+            fullHash = stageInput(staged, input, length, expectedHash, checksum, md5);
+            checkCapacity(bucket, key, length, createOnly);
+            segments = uploadSegments(staged, length);
+        } finally { Files.deleteIfExists(staged); }
+        Metadata metadata = new Metadata(length, Instant.now().toEpochMilli(),
+            HexFormat.of().formatHex(md5.digest()), fullHash, bucket, key, contentType);
+        persistObject(metadata, segments, createOnly);
+        return metadata;
+    }
+
+    private void validatePut(String bucket, long length, String contentType) {
         if (!configuredBucket.equals(bucket)) throw new StoreException(404, "NoSuchBucket", "Bucket not found");
         if (length < 0) throw new StoreException(411, "MissingContentLength", "Content-Length is required");
         if (length > maxObject) throw new StoreException(413, "EntityTooLarge", "Object exceeds the configured size limit");
@@ -59,74 +76,85 @@ final class ClusterStore implements ObjectStorage {
             throw new StoreException(503, "SlowDown", "Fewer than two storage hosts are available");
         if (contentType.getBytes(java.nio.charset.StandardCharsets.UTF_8).length > 255)
             throw new StoreException(400, "InvalidArgument", "Content-Type is too long");
-        MessageDigest sha = digest("SHA-256"), md5 = digest("MD5");
+    }
+
+    private byte[] stageInput(Path staged, InputStream input, long length, String expectedHash,
+                              String checksum, MessageDigest md5) throws IOException {
+        MessageDigest sha = digest("SHA-256");
+        try (OutputStream output = Files.newOutputStream(staged)) {
+            byte[] buffer = new byte[65536];
+            long remaining = length;
+            while (remaining > 0) {
+                int count = input.read(buffer, 0, (int) Math.min(buffer.length, remaining));
+                if (count < 0) throw new StoreException(400, "IncompleteBody", "Payload length does not match Content-Length");
+                if (count == 0) continue;
+                sha.update(buffer, 0, count); md5.update(buffer, 0, count);
+                output.write(buffer, 0, count);
+                remaining -= count;
+            }
+        }
+        if (input.read() != -1) throw new StoreException(413, "EntityTooLarge", "Payload exceeds declared size");
+        byte[] fullHash = sha.digest();
+        if (!HexFormat.of().formatHex(fullHash).equals(expectedHash))
+            throw new StoreException(400, "XAmzContentSHA256Mismatch", "Payload hash mismatch");
+        if (checksum != null && !Base64.getEncoder().encodeToString(fullHash).equals(checksum))
+            throw new StoreException(400, "BadDigest", "SHA-256 checksum mismatch");
+        return fullHash;
+    }
+
+    private void checkCapacity(String bucket, String key, long length, boolean createOnly) throws IOException {
+        try (Connection connection = connect()) {
+            long previous = currentLength(connection, bucket, key);
+            if (createOnly && previous >= 0)
+                throw new StoreException(412, "PreconditionFailed", "Object already exists");
+            try (PreparedStatement query = connection.prepareStatement("SELECT used_bytes FROM cluster_usage WHERE bucket=?")) {
+                query.setString(1, bucket);
+                try (ResultSet result = query.executeQuery()) {
+                    if (!result.next()) throw new SQLException("Bucket quota row is missing");
+                    if (result.getLong(1) - Math.max(0, previous) > maxTotal - length)
+                        throw new StoreException(507, "InsufficientStorage", "Store capacity limit reached");
+                }
+            }
+        } catch (SQLException error) { throw databaseError(error); }
+    }
+
+    private List<Segment> uploadSegments(Path staged, long length) throws IOException {
         List<Segment> segments = new ArrayList<>();
-        byte[] fullHash;
+        try (InputStream stagedInput = Files.newInputStream(staged)) {
-        Path staged = Files.createTempFile("objectstore-cluster-", ".pending");
+            long remaining = length;
-        try {
+            while (remaining > 0) {
-            try (OutputStream output = Files.newOutputStream(staged)) {
+                int wanted = (int) Math.min(ClusterNode.MAX_SEGMENT, remaining);
-                byte[] buffer = new byte[65536];
+                byte[] bytes = stagedInput.readNBytes(wanted);
-                long remaining = length;
+                if (bytes.length != wanted) throw new IOException("Staged object was truncated");
-                while (remaining > 0) {
+                byte[] segmentHash = SigV4.hash(bytes);
-                    int count = input.read(buffer, 0, (int) Math.min(buffer.length, remaining));
+                UUID id = UUID.randomUUID();
-                    if (count < 0) throw new StoreException(400, "IncompleteBody", "Payload length does not match Content-Length");
+                List<UUID> replicas = new ArrayList<>();
-                    if (count == 0) continue;
+                Set<UUID> acceptedHosts = new HashSet<>();
-                    sha.update(buffer, 0, count); md5.update(buffer, 0, count);
+                for (int index : PlacementPolicy.candidates(id, nodes, testNodeDomains)) {
-                    output.write(buffer, 0, count);
+                    UUID host = nodes.faultDomain(index, testNodeDomains);
-                    remaining -= count;
+                    if (acceptedHosts.contains(host)) continue;
-                }
+                    try {
-            }
+                        nodes.put(index, id, bytes, segmentHash);
-            if (input.read() != -1) throw new StoreException(413, "EntityTooLarge", "Payload exceeds declared size");
+                        replicas.add(nodes.node(index).id());
-            fullHash = sha.digest();
+                        acceptedHosts.add(host);
-            if (!HexFormat.of().formatHex(fullHash).equals(expectedHash))
+                        if (acceptedHosts.size() == 3) break;
-                throw new StoreException(400, "XAmzContentSHA256Mismatch", "Payload hash mismatch");
+                    } catch (IOException error) {
-            if (checksum != null && !Base64.getEncoder().encodeToString(fullHash).equals(checksum))
+                        System.err.println("Cluster node " + nodes.node(index).id() +
-                throw new StoreException(400, "BadDigest", "SHA-256 checksum mismatch");
+                            " did not accept segment " + id + ": " + error.getMessage());
-            try (Connection connection = connect()) {
-                long previous = currentLength(connection, bucket, key);
-                if (createOnly && previous >= 0)
-                    throw new StoreException(412, "PreconditionFailed", "Object already exists");
-                try (PreparedStatement query = connection.prepareStatement("SELECT used_bytes FROM cluster_usage WHERE bucket=?")) {
-                    query.setString(1, bucket);
-                    try (ResultSet result = query.executeQuery()) {
-                        if (!result.next()) throw new SQLException("Bucket quota row is missing");
-                        if (result.getLong(1) - Math.max(0, previous) > maxTotal - length)
-                            throw new StoreException(507, "InsufficientStorage", "Store capacity limit reached");
                     }
                 }
-            } catch (SQLException error) { throw databaseError(error); }
+                if (acceptedHosts.size() < 2)
-            try (InputStream stagedInput = Files.newInputStream(staged)) {
+                    throw new StoreException(503, "SlowDown", "Fewer than two storage hosts accepted the segment");
-                long remaining = length;
+                segments.add(new Segment(id, wanted, segmentHash, List.copyOf(replicas)));
-                while (remaining > 0) {
+                remaining -= wanted;
-                    int wanted = (int) Math.min(ClusterNode.MAX_SEGMENT, remaining);
-                    byte[] bytes = stagedInput.readNBytes(wanted);
-                    if (bytes.length != wanted) throw new IOException("Staged object was truncated");
-                    byte[] segmentHash = SigV4.hash(bytes);
-                    UUID id = UUID.randomUUID();
-                    List<UUID> replicas = new ArrayList<>();
-                    Set<UUID> acceptedHosts = new HashSet<>();
-                    for (int index : PlacementPolicy.candidates(id, nodes, testNodeDomains)) {
-                        UUID host = nodes.faultDomain(index, testNodeDomains);
-                        if (acceptedHosts.contains(host)) continue;
-                        try {
-                            nodes.put(index, id, bytes, segmentHash);
-                            replicas.add(nodes.node(index).id());
-                            acceptedHosts.add(host);
-                            if (acceptedHosts.size() == 3) break;
-                        } catch (IOException error) {
-                            System.err.println("Cluster node " + nodes.node(index).id() +
-                                " did not accept segment " + id + ": " + error.getMessage());
-                        }
-                    }
-                    if (acceptedHosts.size() < 2)
-                        throw new StoreException(503, "SlowDown", "Fewer than two storage hosts accepted the segment");
-                    segments.add(new Segment(id, wanted, segmentHash, List.copyOf(replicas)));
-                    remaining -= wanted;
-                }
             }
-        } finally { Files.deleteIfExists(staged); }
+        }
-        Metadata metadata = new Metadata(length, Instant.now().toEpochMilli(),
+        return segments;
-            HexFormat.of().formatHex(md5.digest()), fullHash, bucket, key, contentType);
+    }
+
+    private void persistObject(Metadata metadata, List<Segment> segments, boolean createOnly) throws IOException {
+        String bucket = metadata.bucket(), key = metadata.key();
+        long length = metadata.length();
         UUID generation = UUID.randomUUID();
         try (Connection connection = connect()) {
             connection.setAutoCommit(false);
@@ -159,7 +187,6 @@ final class ClusterStore implements ObjectStorage {
                     delete.setString(1, bucket); delete.setString(2, key); delete.executeUpdate();
                 }
                 connection.commit();
-                return metadata;
             } catch (SQLException | RuntimeException error) {
                 connection.rollback();
                 if (error instanceof SQLException sql) throw databaseError(sql);
@@ -241,45 +268,54 @@ final class ClusterStore implements ObjectStorage {
     }
 
     @Override public ListPage list(String bucket, String prefix, String delimiter, int maxKeys, String after) throws IOException {
-        List<ListedObject> entries = new ArrayList<>();
+        if (maxKeys == 0) return new ListPage(new ArrayList<>(), new ArrayList<>(), null, false);
-        List<String> prefixes = new ArrayList<>();
-        if (maxKeys == 0) return new ListPage(entries, prefixes, null, false);
-        String lastKey = null, activePrefix = null;
-        boolean truncated = false;
         try (Connection connection = connect()) {
             connection.setAutoCommit(false);
+            ListPage page;
             try (PreparedStatement query = connection.prepareStatement(
                 "SELECT object_key, length, modified, etag, sha256, content_type FROM cluster_objects WHERE bucket=? AND object_key>=? ORDER BY object_key")) {
                 query.setString(1, bucket);
                 query.setString(2, after != null && after.compareTo(prefix) > 0 ? after : prefix);
                 query.setFetchSize(128);
                 try (ResultSet result = query.executeQuery()) {
-                    while (result.next()) {
+                    page = readListPage(result, bucket, prefix, delimiter, maxKeys, after);
-                        String key = result.getString(1);
-                        if (!key.startsWith(prefix)) break;
-                        if (after != null && key.compareTo(after) <= 0) continue;
-                        String group = null;
-                        if (!delimiter.isEmpty()) {
-                            int at = key.indexOf(delimiter, prefix.length());
-                            if (at >= 0) group = key.substring(0, at + delimiter.length());
-                        }
-                        if (group != null && group.equals(activePrefix)) { lastKey = key; continue; }
-                        if (entries.size() + prefixes.size() >= maxKeys) { truncated = true; break; }
-                        if (group != null) { prefixes.add(group); activePrefix = group; }
-                        else {
-                            entries.add(new ListedObject(key, new Metadata(result.getLong(2), result.getLong(3),
-                                result.getString(4), result.getBytes(5), bucket, key, result.getString(6))));
-                            activePrefix = null;
-                        }
-                        lastKey = key;
-                    }
                 }
             }
             connection.commit();
+            return page;
         } catch (SQLException error) { throw databaseError(error); }
+    }
+
+    private static ListPage readListPage(ResultSet result, String bucket, String prefix, String delimiter,
+                                         int maxKeys, String after) throws SQLException {
+        List<ListedObject> entries = new ArrayList<>();
+        List<String> prefixes = new ArrayList<>();
+        String lastKey = null, activePrefix = null;
+        boolean truncated = false;
+        while (result.next()) {
+            String key = result.getString(1);
+            if (!key.startsWith(prefix)) break;
+            if (after != null && key.compareTo(after) <= 0) continue;
+            String group = commonPrefix(key, prefix, delimiter);
+            if (group != null && group.equals(activePrefix)) { lastKey = key; continue; }
+            if (entries.size() + prefixes.size() >= maxKeys) { truncated = true; break; }
+            if (group != null) { prefixes.add(group); activePrefix = group; }
+            else {
+                entries.add(new ListedObject(key, new Metadata(result.getLong(2), result.getLong(3),
+                    result.getString(4), result.getBytes(5), bucket, key, result.getString(6))));
+                activePrefix = null;
+            }
+            lastKey = key;
+        }
         return new ListPage(entries, prefixes, truncated ? lastKey : null, truncated);
     }
 
+    private static String commonPrefix(String key, String prefix, String delimiter) {
+        if (delimiter.isEmpty()) return null;
+        int at = key.indexOf(delimiter, prefix.length());
+        return at < 0 ? null : key.substring(0, at + delimiter.length());
+    }
+
     private long lockUsage(Connection connection, String bucket) throws SQLException {
         try (PreparedStatement query = connection.prepareStatement("SELECT used_bytes FROM cluster_usage WHERE bucket=? FOR UPDATE")) {
             query.setString(1, bucket);
@@ -364,23 +400,21 @@ final class ClusterStore implements ObjectStorage {
                         for (UUID id : segment.replicas()) {
                             int node = nodes.index(id);
                             if (node < 0) continue;
-                            try {
+                            byte[] candidate = readableReplica(node, segment);
-                                byte[] candidate = nodes.get(node, segment.id(), segment.length(), segment.hash());
+                            if (candidate == null) continue;
-                                if (copy == null) copy = candidate;
+                            if (copy == null) copy = candidate;
-                                healthy.add(id);
+                            healthy.add(id);
-                                healthyHosts.add(nodes.faultDomain(node, testNodeDomains));
+                            healthyHosts.add(nodes.faultDomain(node, testNodeDomains));
-                            } catch (IOException error) { }
                         }
                         if (copy == null) { unrecoverable++; continue; }
                         for (int node : PlacementPolicy.candidates(segment.id(), nodes, testNodeDomains)) {
                             UUID host = nodes.faultDomain(node, testNodeDomains);
                             if (healthyHosts.contains(host)) continue;
-                            try {
+                            if (repairReplica(node, segment, copy)) {
-                                nodes.repair(node, segment.id(), copy, segment.hash());
                                 healthy.add(nodes.node(node).id());
                                 healthyHosts.add(host);
                                 restored++;
-                            } catch (IOException error) { }
+                            }
                             if (healthyHosts.size() == 3) break;
                         }
                         if (healthyHosts.size() < 3) underReplicated++;
@@ -403,6 +437,19 @@ final class ClusterStore implements ObjectStorage {
         } catch (SQLException error) { throw databaseError(error); }
         return new RepairReport(scanned, restored, underReplicated, unrecoverable);
     }
+
+    private byte[] readableReplica(int node, Segment segment) {
+        try { return nodes.get(node, segment.id(), segment.length(), segment.hash()); }
+        catch (IOException unavailable) { return null; }
+    }
+
+    private boolean repairReplica(int node, Segment segment, byte[] copy) {
+        try {
+            nodes.repair(node, segment.id(), copy, segment.hash());
+            return true;
+        } catch (IOException unavailable) { return false; }
+    }
+
     @Override public void close() {}
 
     private final class SegmentStream extends InputStream {

src/cloud/lunarsky/store/DiskStore.java

@@ -117,67 +117,82 @@ final class DiskStore implements ObjectStorage {
         byte[] bucketBytes = bucket.getBytes(StandardCharsets.UTF_8);
         byte[] keyBytes = key.getBytes(StandardCharsets.UTF_8);
         byte[] typeBytes = contentType.getBytes(StandardCharsets.UTF_8);
-        if (bucketBytes.length > 63 || keyBytes.length > 1024 || typeBytes.length > 255)
+        validateMetadataLengths(bucketBytes, keyBytes, typeBytes);
-            throw new StoreException(400, "InvalidArgument", "Object metadata is too long");
-        int headerLength = HEADER_V2 + bucketBytes.length + keyBytes.length + typeBytes.length;
         Path destination = object(bucket, key), pending = Files.createTempFile(temporary, "upload-", ".part");
         try {
-            MessageDigest sha = digest("SHA-256"), md5 = digest("MD5");
+            Metadata metadata = stagePut(pending, input, length, expectedHash, checksum,
-            long count = 0;
+                bucket, key, contentType, bucketBytes, keyBytes, typeBytes);
-            try (OutputStream out = Files.newOutputStream(pending)) {
+            installPending(destination, pending, metadata, createOnly);
-                out.write(new byte[headerLength]);
-                byte[] buffer = new byte[65536]; int n;
-                while ((n = input.read(buffer)) != -1) {
-                    count += n;
-                    if (count > length || count > maxObject)
-                        throw new StoreException(413, "EntityTooLarge", "Payload exceeds declared size");
-                    sha.update(buffer, 0, n); md5.update(buffer, 0, n); out.write(buffer, 0, n);
-                }
-            }
-            if (count != length) throw new StoreException(400, "IncompleteBody", "Payload length does not match Content-Length");
-            byte[] hash = sha.digest(), etag = md5.digest();
-            if (!MessageDigest.isEqual(hash, HexFormat.of().parseHex(expectedHash)))
-                throw new StoreException(400, "XAmzContentSHA256Mismatch", "Payload hash mismatch");
-            if (checksum != null && !Base64.getEncoder().encodeToString(hash).equals(checksum))
-                throw new StoreException(400, "BadDigest", "SHA-256 checksum mismatch");
-            long modified = Instant.now().toEpochMilli();
-            ByteBuffer header = ByteBuffer.allocate(headerLength).putLong(MAGIC_V2).putLong(count)
-                .putLong(modified).put(etag).put(hash).putShort((short) bucketBytes.length)
-                .putShort((short) keyBytes.length).putShort((short) typeBytes.length)
-                .put(bucketBytes).put(keyBytes).put(typeBytes);
-            header.flip();
-            try (FileChannel file = FileChannel.open(pending, StandardOpenOption.WRITE)) {
-                while (header.hasRemaining()) file.write(header, header.position());
-                file.force(true);
-            }
-            Metadata metadata = new Metadata(count, modified, SigV4.hex(etag), hash, bucket, key, contentType);
-            synchronized (lock(destination)) {
-                long previous = 0;
-                boolean existed = Files.exists(destination);
-                boolean legacy = false;
-                if (existed) {
-                    if (createOnly) throw new StoreException(412, "PreconditionFailed", "Object already exists");
-                    try (var in = new DataInputStream(Files.newInputStream(destination))) {
-                        Metadata old = readRecord(in).metadata();
-                        previous = old.length();
-                        legacy = old.key() == null;
-                    }
-                }
-                synchronized (this) {
-                    if (used - previous + count > maxTotal)
-                        throw new StoreException(507, "InsufficientStorage", "Store capacity limit reached");
-                    Files.move(pending, destination, StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING);
-                    used = used - previous + count;
-                    if (!existed) objectCount++;
-                    if (legacy) legacyCount--;
-                    index.put(indexKey(bucket, key), metadata);
-                    syncDirectory(destination.getParent());
-                }
-            }
             return metadata;
         } finally { Files.deleteIfExists(pending); }
     }
 
+    private static void validateMetadataLengths(byte[] bucketBytes, byte[] keyBytes, byte[] typeBytes) {
+        if (bucketBytes.length > 63 || keyBytes.length > 1024 || typeBytes.length > 255)
+            throw new StoreException(400, "InvalidArgument", "Object metadata is too long");
+    }
+
+    private Metadata stagePut(Path pending, InputStream input, long length, String expectedHash, String checksum,
+                              String bucket, String key, String contentType,
+                              byte[] bucketBytes, byte[] keyBytes, byte[] typeBytes) throws IOException {
+        int headerLength = HEADER_V2 + bucketBytes.length + keyBytes.length + typeBytes.length;
+        MessageDigest sha = digest("SHA-256"), md5 = digest("MD5");
+        long count = 0;
+        try (OutputStream out = Files.newOutputStream(pending)) {
+            out.write(new byte[headerLength]);
+            byte[] buffer = new byte[65536]; int n;
+            while ((n = input.read(buffer)) != -1) {
+                count += n;
+                if (count > length || count > maxObject)
+                    throw new StoreException(413, "EntityTooLarge", "Payload exceeds declared size");
+                sha.update(buffer, 0, n); md5.update(buffer, 0, n); out.write(buffer, 0, n);
+            }
+        }
+        if (count != length) throw new StoreException(400, "IncompleteBody", "Payload length does not match Content-Length");
+        byte[] hash = sha.digest(), etag = md5.digest();
+        if (!MessageDigest.isEqual(hash, HexFormat.of().parseHex(expectedHash)))
+            throw new StoreException(400, "XAmzContentSHA256Mismatch", "Payload hash mismatch");
+        if (checksum != null && !Base64.getEncoder().encodeToString(hash).equals(checksum))
+            throw new StoreException(400, "BadDigest", "SHA-256 checksum mismatch");
+        long modified = Instant.now().toEpochMilli();
+        ByteBuffer header = ByteBuffer.allocate(headerLength).putLong(MAGIC_V2).putLong(count)
+            .putLong(modified).put(etag).put(hash).putShort((short) bucketBytes.length)
+            .putShort((short) keyBytes.length).putShort((short) typeBytes.length)
+            .put(bucketBytes).put(keyBytes).put(typeBytes);
+        header.flip();
+        try (FileChannel file = FileChannel.open(pending, StandardOpenOption.WRITE)) {
+            while (header.hasRemaining()) file.write(header, header.position());
+            file.force(true);
+        }
+        return new Metadata(count, modified, SigV4.hex(etag), hash, bucket, key, contentType);
+    }
+
+    private void installPending(Path destination, Path pending, Metadata metadata, boolean createOnly) throws IOException {
+        synchronized (lock(destination)) {
+            long previous = 0;
+            boolean existed = Files.exists(destination);
+            boolean legacy = false;
+            if (existed) {
+                if (createOnly) throw new StoreException(412, "PreconditionFailed", "Object already exists");
+                try (var in = new DataInputStream(Files.newInputStream(destination))) {
+                    Metadata old = readRecord(in).metadata();
+                    previous = old.length();
+                    legacy = old.key() == null;
+                }
+            }
+            synchronized (this) {
+                if (used - previous + metadata.length() > maxTotal)
+                    throw new StoreException(507, "InsufficientStorage", "Store capacity limit reached");
+                Files.move(pending, destination, StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING);
+                used = used - previous + metadata.length();
+                if (!existed) objectCount++;
+                if (legacy) legacyCount--;
+                index.put(indexKey(metadata.bucket(), metadata.key()), metadata);
+                syncDirectory(destination.getParent());
+            }
+        }
+    }
+
     public OpenObject open(String bucket, String key) throws IOException {
         Path destination = object(bucket, key);
         synchronized (lock(destination)) {

src/cloud/lunarsky/store/Main.java

@@ -45,87 +45,14 @@ public final class Main {
         exchange.getResponseHeaders().set("X-Content-Type-Options", "nosniff");
         try {
             if (!admitted) throw new StoreException(503, "SlowDown", "Too many concurrent requests");
-            if (exchange.getRequestURI().getRawPath().equals("/health") && exchange.getRequestMethod().equals("GET")) {
+            if (handleStatus(exchange)) return;
-                byte[] body = "{\"status\":\"ok\",\"service\":\"lunarsky-objectstore\"}".getBytes(StandardCharsets.UTF_8);
-                exchange.getResponseHeaders().set("Content-Type", "application/json");
-                exchange.sendResponseHeaders(200, body.length);
-                exchange.getResponseBody().write(body);
-                return;
-            }
-            if (exchange.getRequestURI().getRawPath().equals("/ready") && exchange.getRequestMethod().equals("GET")) {
-                boolean ready = store.ready();
-                byte[] body = (ready ? "ready" : "unavailable").getBytes(StandardCharsets.UTF_8);
-                exchange.getResponseHeaders().set("Content-Type", "text/plain; charset=utf-8");
-                exchange.sendResponseHeaders(ready ? 200 : 503, body.length);
-                exchange.getResponseBody().write(body);
-                return;
-            }
             String hash = authentication.verify(exchange.getRequestMethod(), exchange.getRequestURI(), exchange.getRequestHeaders());
             String path = SigV4.decode(exchange.getRequestURI().getRawPath());
             Map<String, String> query = query(exchange.getRequestURI().getRawQuery());
             if (path.equals("/" + bucket) || path.equals("/" + bucket + "/")) {
-                if (!exchange.getRequestMethod().equals("GET") || !"2".equals(query.get("list-type")) ||
+                handleBucket(exchange, query, hash);
-                    !query.keySet().stream().allMatch(java.util.Set.of("list-type", "prefix", "delimiter", "max-keys",
+            } else {
-                        "continuation-token", "start-after", "encoding-type", "x-id")::contains) ||
+                handleObject(exchange, path, query, hash);
-                    (query.containsKey("x-id") && !"ListObjectsV2".equals(query.get("x-id"))))
-                    unsupported("Bucket operation");
-                requireEmptyBody(exchange, hash);
-                listObjects(exchange, query);
-                return;
-            }
-            String prefix = "/" + bucket + "/";
-            if (!path.startsWith(prefix)) throw new StoreException(404, "NoSuchBucket", "Bucket not found");
-            String key = path.substring(prefix.length());
-            if (key.isEmpty() || key.getBytes(StandardCharsets.UTF_8).length > 1024 || key.indexOf('\0') >= 0)
-                throw new StoreException(400, "InvalidArgument", "Invalid object key");
-            String method = exchange.getRequestMethod();
-            boolean multipartRequest = multipartRequest(method, query);
-            if (!multipartRequest && !query.isEmpty() && !(query.size() == 1 &&
-                ("PutObject".equals(query.get("x-id")) || "GetObject".equals(query.get("x-id")) ||
-                 "HeadObject".equals(query.get("x-id")) || "DeleteObject".equals(query.get("x-id")))))
-                unsupported("Query operation");
-            var headers = exchange.getRequestHeaders();
-            for (String name : headers.keySet()) {
-                String lower = name.toLowerCase(java.util.Locale.ROOT);
-                if (lower.startsWith("x-amz-") && !java.util.Set.of("x-amz-date", "x-amz-content-sha256",
-                    "x-amz-checksum-sha256", "x-amz-sdk-checksum-algorithm", "x-amz-user-agent").contains(lower))
-                    unsupported("Amazon header");
-                if (lower.startsWith("x-amz-meta-") || lower.startsWith("x-amz-server-side-") ||
-                    lower.startsWith("x-amz-copy-") || lower.startsWith("x-amz-acl") ||
-                    lower.startsWith("x-amz-grant") || lower.startsWith("x-amz-tagging") ||
-                    lower.equals("content-md5")) unsupported("Object metadata, encryption, ACL, copy, tagging or MD5 header");
-                if (lower.startsWith("x-amz-checksum-") && !lower.equals("x-amz-checksum-sha256"))
-                    unsupported("Checksum algorithm");
-            }
-            String algorithm = SigV4.single(headers, "x-amz-sdk-checksum-algorithm");
-            if (algorithm != null && !algorithm.equals("SHA256")) unsupported("Checksum algorithm");
-            if (multipartRequest) {
-                handleMultipart(exchange, method, query, key, hash);
-                return;
-            }
-            if (!method.equals("PUT")) requireEmptyBody(exchange, hash);
-            switch (method) {
-                case "PUT" -> {
-                    String length = SigV4.single(headers, "content-length"), condition = SigV4.single(headers, "if-none-match");
-                    if (condition != null && !condition.equals("*")) unsupported("Write condition");
-                    long bytes;
-                    try { bytes = length == null ? -1 : Long.parseLong(length); }
-                    catch (NumberFormatException e) { throw new StoreException(400, "InvalidArgument", "Invalid Content-Length"); }
-                    if (headers.containsKey("content-encoding")) unsupported("Encoded payload");
-                    String contentType = contentType(headers);
-                    ObjectStorage.Metadata data = store.put(bucket, key, exchange.getRequestBody(), bytes, hash,
-                        SigV4.single(headers, "x-amz-checksum-sha256"), condition != null, contentType);
-                    exchange.getResponseHeaders().set("ETag", "\"" + data.etag() + "\"");
-                    exchange.getResponseHeaders().set("x-amz-checksum-sha256", Base64.getEncoder().encodeToString(data.sha256()));
-                    exchange.sendResponseHeaders(200, -1);
-                }
-                case "GET", "HEAD" -> readObject(exchange, key);
-                case "DELETE" -> {
-                    if (headers.containsKey("if-none-match")) unsupported("Conditional delete");
-                    store.delete(bucket, key);
-                    exchange.sendResponseHeaders(204, -1);
-                }
-                default -> unsupported("HTTP method");
             }
         } catch (StoreException error) { sendError(exchange, error.status, error.code, error.getMessage(), requestId); }
         catch (Exception error) {
@@ -134,6 +61,100 @@ public final class Main {
         } finally { if (admitted) slots.release(); exchange.close(); }
     }
 
+    private boolean handleStatus(HttpExchange exchange) throws IOException {
+        if (!exchange.getRequestMethod().equals("GET")) return false;
+        String path = exchange.getRequestURI().getRawPath();
+        if (path.equals("/health")) {
+            byte[] body = "{\"status\":\"ok\",\"service\":\"lunarsky-objectstore\"}".getBytes(StandardCharsets.UTF_8);
+            exchange.getResponseHeaders().set("Content-Type", "application/json");
+            exchange.sendResponseHeaders(200, body.length);
+            exchange.getResponseBody().write(body);
+            return true;
+        }
+        if (!path.equals("/ready")) return false;
+        boolean ready = store.ready();
+        byte[] body = (ready ? "ready" : "unavailable").getBytes(StandardCharsets.UTF_8);
+        exchange.getResponseHeaders().set("Content-Type", "text/plain; charset=utf-8");
+        exchange.sendResponseHeaders(ready ? 200 : 503, body.length);
+        exchange.getResponseBody().write(body);
+        return true;
+    }
+
+    private void handleBucket(HttpExchange exchange, Map<String, String> query, String hash) throws IOException {
+        if (!exchange.getRequestMethod().equals("GET") || !"2".equals(query.get("list-type")) ||
+            !query.keySet().stream().allMatch(java.util.Set.of("list-type", "prefix", "delimiter", "max-keys",
+                "continuation-token", "start-after", "encoding-type", "x-id")::contains) ||
+            (query.containsKey("x-id") && !"ListObjectsV2".equals(query.get("x-id"))))
+            unsupported("Bucket operation");
+        requireEmptyBody(exchange, hash);
+        listObjects(exchange, query);
+    }
+
+    private void handleObject(HttpExchange exchange, String path, Map<String, String> query, String hash) throws IOException {
+        String prefix = "/" + bucket + "/";
+        if (!path.startsWith(prefix)) throw new StoreException(404, "NoSuchBucket", "Bucket not found");
+        String key = path.substring(prefix.length());
+        if (key.isEmpty() || key.getBytes(StandardCharsets.UTF_8).length > 1024 || key.indexOf('\0') >= 0)
+            throw new StoreException(400, "InvalidArgument", "Invalid object key");
+        String method = exchange.getRequestMethod();
+        boolean multipartRequest = multipartRequest(method, query);
+        if (!multipartRequest && !query.isEmpty() && !(query.size() == 1 &&
+            ("PutObject".equals(query.get("x-id")) || "GetObject".equals(query.get("x-id")) ||
+             "HeadObject".equals(query.get("x-id")) || "DeleteObject".equals(query.get("x-id")))))
+            unsupported("Query operation");
+        validateObjectHeaders(exchange.getRequestHeaders());
+        if (multipartRequest) {
+            handleMultipart(exchange, method, query, key, hash);
+            return;
+        }
+        if (!method.equals("PUT")) requireEmptyBody(exchange, hash);
+        switch (method) {
+            case "PUT" -> putObject(exchange, key, hash);
+            case "GET", "HEAD" -> readObject(exchange, key);
+            case "DELETE" -> deleteObject(exchange, key);
+            default -> unsupported("HTTP method");
+        }
+    }
+
+    private static void validateObjectHeaders(com.sun.net.httpserver.Headers headers) {
+        for (String name : headers.keySet()) {
+            String lower = name.toLowerCase(java.util.Locale.ROOT);
+            if (lower.startsWith("x-amz-") && !java.util.Set.of("x-amz-date", "x-amz-content-sha256",
+                "x-amz-checksum-sha256", "x-amz-sdk-checksum-algorithm", "x-amz-user-agent").contains(lower))
+                unsupported("Amazon header");
+            if (lower.startsWith("x-amz-meta-") || lower.startsWith("x-amz-server-side-") ||
+                lower.startsWith("x-amz-copy-") || lower.startsWith("x-amz-acl") ||
+                lower.startsWith("x-amz-grant") || lower.startsWith("x-amz-tagging") ||
+                lower.equals("content-md5")) unsupported("Object metadata, encryption, ACL, copy, tagging or MD5 header");
+            if (lower.startsWith("x-amz-checksum-") && !lower.equals("x-amz-checksum-sha256"))
+                unsupported("Checksum algorithm");
+        }
+        String algorithm = SigV4.single(headers, "x-amz-sdk-checksum-algorithm");
+        if (algorithm != null && !algorithm.equals("SHA256")) unsupported("Checksum algorithm");
+    }
+
+    private void putObject(HttpExchange exchange, String key, String hash) throws IOException {
+        var headers = exchange.getRequestHeaders();
+        String length = SigV4.single(headers, "content-length"), condition = SigV4.single(headers, "if-none-match");
+        if (condition != null && !condition.equals("*")) unsupported("Write condition");
+        long bytes;
+        try { bytes = length == null ? -1 : Long.parseLong(length); }
+        catch (NumberFormatException e) { throw new StoreException(400, "InvalidArgument", "Invalid Content-Length"); }
+        if (headers.containsKey("content-encoding")) unsupported("Encoded payload");
+        String type = contentType(headers);
+        ObjectStorage.Metadata data = store.put(bucket, key, exchange.getRequestBody(), bytes, hash,
+            SigV4.single(headers, "x-amz-checksum-sha256"), condition != null, type);
+        exchange.getResponseHeaders().set("ETag", "\"" + data.etag() + "\"");
+        exchange.getResponseHeaders().set("x-amz-checksum-sha256", Base64.getEncoder().encodeToString(data.sha256()));
+        exchange.sendResponseHeaders(200, -1);
+    }
+
+    private void deleteObject(HttpExchange exchange, String key) throws IOException {
+        if (exchange.getRequestHeaders().containsKey("if-none-match")) unsupported("Conditional delete");
+        store.delete(bucket, key);
+        exchange.sendResponseHeaders(204, -1);
+    }
+
     private static String contentType(com.sun.net.httpserver.Headers headers) {
         String value = SigV4.single(headers, "content-type");
         if (value == null) return "application/octet-stream";
@@ -337,6 +358,24 @@ public final class Main {
     }
 
     private void listObjects(HttpExchange exchange, Map<String, String> query) throws IOException {
+        ListRequest request = listRequest(query);
+        var page = store.list(bucket, request.prefix(), request.delimiter(), request.maxKeys(), request.after());
+        StringBuilder xml = new StringBuilder("<?xml version=\"1.0\" encoding=\"UTF-8\"?><ListBucketResult xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\">");
+        appendListHeader(xml, query, request, page);
+        appendListEntries(xml, page, request.encoding());
+        if (page.truncated()) xml.append("<NextContinuationToken>")
+            .append(Base64.getUrlEncoder().withoutPadding().encodeToString(page.nextKey().getBytes(StandardCharsets.UTF_8)))
+            .append("</NextContinuationToken>");
+        xml.append("</ListBucketResult>");
+        byte[] body = xml.toString().getBytes(StandardCharsets.UTF_8);
+        exchange.getResponseHeaders().set("Content-Type", "application/xml");
+        exchange.sendResponseHeaders(200, body.length);
+        exchange.getResponseBody().write(body);
+    }
+
+    private record ListRequest(String prefix, String delimiter, String encoding, int maxKeys, String after) { }
+
+    private static ListRequest listRequest(Map<String, String> query) {
         String prefix = query.getOrDefault("prefix", ""), delimiter = query.getOrDefault("delimiter", "");
         String encoding = query.get("encoding-type");
         if (encoding != null && !encoding.equals("url")) unsupported("Encoding type");
@@ -356,8 +395,11 @@ public final class Main {
                 throw new StoreException(400, "InvalidArgument", "Invalid continuation token");
             }
         }
-        var page = store.list(bucket, prefix, delimiter, maxKeys, after);
+        return new ListRequest(prefix, delimiter, encoding, maxKeys, after);
-        StringBuilder xml = new StringBuilder("<?xml version=\"1.0\" encoding=\"UTF-8\"?><ListBucketResult xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\">");
+    }
+
+    private void appendListHeader(StringBuilder xml, Map<String, String> query, ListRequest request, ObjectStorage.ListPage page) {
+        String prefix = request.prefix(), delimiter = request.delimiter(), encoding = request.encoding();
         xml.append("<Name>").append(xml(bucket)).append("</Name><Prefix>").append(xml(listKey(prefix, encoding))).append("</Prefix>");
         if (!delimiter.isEmpty()) xml.append("<Delimiter>").append(xml(listKey(delimiter, encoding))).append("</Delimiter>");
         if (encoding != null) xml.append("<EncodingType>url</EncodingType>");
@@ -365,8 +407,11 @@ public final class Main {
             .append(xml(query.get("continuation-token"))).append("</ContinuationToken>");
         if (query.containsKey("start-after")) xml.append("<StartAfter>")
             .append(xml(listKey(query.get("start-after"), encoding))).append("</StartAfter>");
-        xml.append("<KeyCount>").append(page.keyCount()).append("</KeyCount><MaxKeys>").append(maxKeys)
+        xml.append("<KeyCount>").append(page.keyCount()).append("</KeyCount><MaxKeys>").append(request.maxKeys())
             .append("</MaxKeys><IsTruncated>").append(page.truncated()).append("</IsTruncated>");
+    }
+
+    private static void appendListEntries(StringBuilder xml, ObjectStorage.ListPage page, String encoding) {
         int objectAt = 0, prefixAt = 0;
         while (objectAt < page.objects().size() || prefixAt < page.prefixes().size()) {
             if (objectAt < page.objects().size() &&
@@ -384,14 +429,6 @@ public final class Main {
                     .append("</Prefix></CommonPrefixes>");
             }
         }
-        if (page.truncated()) xml.append("<NextContinuationToken>")
-            .append(Base64.getUrlEncoder().withoutPadding().encodeToString(page.nextKey().getBytes(StandardCharsets.UTF_8)))
-            .append("</NextContinuationToken>");
-        xml.append("</ListBucketResult>");
-        byte[] body = xml.toString().getBytes(StandardCharsets.UTF_8);
-        exchange.getResponseHeaders().set("Content-Type", "application/xml");
-        exchange.sendResponseHeaders(200, body.length);
-        exchange.getResponseBody().write(body);
     }
 
     private static String listKey(String key, String encoding) {

src/cloud/lunarsky/store/NodeClient.java

@@ -74,6 +74,11 @@ final class NodeClient {
         }
     }
 
+    static NodeIdentity probeIfAvailable(URI url, String token) {
+        try { return probe(url, token); }
+        catch (IOException offline) { return null; }
+    }
+
     static void validateUrl(URI url) {
         if (url == null || !"http".equals(url.getScheme()) || url.getHost() == null ||
             url.getPort() < 1 || url.getRawUserInfo() != null ||
@@ -86,12 +91,11 @@ final class NodeClient {
         Set<UUID> healthy = new HashSet<>();
         for (int i = 0; i < nodes.size(); i++) {
             Node node = nodes.get(i);
-            try {
+            NodeIdentity actual = probeIfAvailable(node.url(), token);
-                NodeIdentity actual = probe(node.url(), token);
+            if (actual == null) continue;
-                if (actual.nodeId().equals(node.id()) && actual.hostId().equals(node.hostId()))
+            if (actual.nodeId().equals(node.id()) && actual.hostId().equals(node.hostId()))
-                    healthy.add(faultDomain(i, testNodeDomains));
+                healthy.add(faultDomain(i, testNodeDomains));
-                if (healthy.size() >= required) return true;
+            if (healthy.size() >= required) return true;
-            } catch (IOException error) { }
         }
         return false;
     }

src/cloud/lunarsky/store/NodeRegistry.java

@@ -69,56 +69,10 @@ final class NodeRegistry {
             try (Statement statement = connection.createStatement()) {
                 statement.execute("SELECT pg_advisory_xact_lock(6834071092781)");
             }
-            Map<String, NodeClient.Node> stored = new HashMap<>();
+            Map<String, NodeClient.Node> stored = registeredNodes(connection);
-            try (Statement statement = connection.createStatement();
+            if (stored.isEmpty()) registerInitialNodes(connection, urls, token, stored);
-                 ResultSet result = statement.executeQuery("SELECT node_id, host_id, endpoint FROM cluster_nodes WHERE state <> 'retired'")) {
+            List<NodeClient.Node> configured = configuredNodes(urls, token, stored);
-                while (result.next()) {
+            ensureLiveReplicasConfigured(connection, configured);
-                    URI url = URI.create(result.getString(3));
-                    stored.put(url.toString(), new NodeClient.Node((UUID) result.getObject(1),
-                        (UUID) result.getObject(2), url));
-                }
-            }
-            if (stored.isEmpty()) {
-                try (Statement statement = connection.createStatement();
-                     ResultSet result = statement.executeQuery("SELECT EXISTS (SELECT 1 FROM cluster_segments)")) {
-                    result.next();
-                    if (result.getBoolean(1)) throw new IOException("Existing segments have no registered node identities");
-                }
-                for (URI url : urls) {
-                    NodeIdentity identity = NodeClient.probe(url, token);
-                    try (PreparedStatement insert = connection.prepareStatement(
-                            "INSERT INTO cluster_nodes (node_id, host_id, endpoint, state) VALUES (?, ?, ?, 'active')")) {
-                        insert.setObject(1, identity.nodeId());
-                        insert.setObject(2, identity.hostId());
-                        insert.setString(3, url.toString());
-                        insert.executeUpdate();
-                    }
-                    stored.put(url.toString(), new NodeClient.Node(identity.nodeId(), identity.hostId(), url));
-                }
-            }
-            List<NodeClient.Node> configured = new ArrayList<>();
-            Set<UUID> configuredIds = new HashSet<>();
-            for (URI url : urls) {
-                NodeClient.Node node = stored.get(url.toString());
-                if (node == null) throw new IOException("Unregistered storage node URL: " + url);
-                NodeIdentity actual = null;
-                try {
-                    actual = NodeClient.probe(url, token);
-                } catch (IOException offline) { }
-                if (actual != null && (!actual.nodeId().equals(node.id()) || !actual.hostId().equals(node.hostId())))
-                    throw new IOException("Storage node identity changed at " + url);
-                configured.add(node);
-                configuredIds.add(node.id());
-            }
-            try (Statement statement = connection.createStatement();
-                 ResultSet result = statement.executeQuery(
-                     "SELECT DISTINCT unnest(s.replica_ids) FROM cluster_segments s JOIN cluster_objects o ON o.generation=s.generation")) {
-                while (result.next()) {
-                    UUID id = (UUID) result.getObject(1);
-                    if (!configuredIds.contains(id))
-                        throw new IOException("A live segment refers to a node missing from CLUSTER_NODES: " + id);
-                }
-            }
             NodeClient nodes = new NodeClient(configured, token, repairToken);
             connection.commit();
             return nodes;
@@ -132,4 +86,66 @@ final class NodeRegistry {
             catch (SQLException error) { throw new IOException("Could not restore metadata connection", error); }
         }
     }
+
+    private static Map<String, NodeClient.Node> registeredNodes(Connection connection) throws SQLException {
+        Map<String, NodeClient.Node> stored = new HashMap<>();
+        try (Statement statement = connection.createStatement();
+             ResultSet result = statement.executeQuery("SELECT node_id, host_id, endpoint FROM cluster_nodes WHERE state <> 'retired'")) {
+            while (result.next()) {
+                URI url = URI.create(result.getString(3));
+                stored.put(url.toString(), new NodeClient.Node((UUID) result.getObject(1),
+                    (UUID) result.getObject(2), url));
+            }
+        }
+        return stored;
+    }
+
+    private static void registerInitialNodes(Connection connection, List<URI> urls, String token,
+                                             Map<String, NodeClient.Node> stored) throws SQLException, IOException {
+        try (Statement statement = connection.createStatement();
+             ResultSet result = statement.executeQuery("SELECT EXISTS (SELECT 1 FROM cluster_segments)")) {
+            result.next();
+            if (result.getBoolean(1)) throw new IOException("Existing segments have no registered node identities");
+        }
+        for (URI url : urls) {
+            NodeIdentity identity = NodeClient.probe(url, token);
+            try (PreparedStatement insert = connection.prepareStatement(
+                    "INSERT INTO cluster_nodes (node_id, host_id, endpoint, state) VALUES (?, ?, ?, 'active')")) {
+                insert.setObject(1, identity.nodeId());
+                insert.setObject(2, identity.hostId());
+                insert.setString(3, url.toString());
+                insert.executeUpdate();
+            }
+            stored.put(url.toString(), new NodeClient.Node(identity.nodeId(), identity.hostId(), url));
+        }
+    }
+
+    private static List<NodeClient.Node> configuredNodes(List<URI> urls, String token,
+                                                         Map<String, NodeClient.Node> stored) throws IOException {
+        List<NodeClient.Node> configured = new ArrayList<>();
+        for (URI url : urls) {
+            NodeClient.Node node = stored.get(url.toString());
+            if (node == null) throw new IOException("Unregistered storage node URL: " + url);
+            NodeIdentity actual = NodeClient.probeIfAvailable(url, token);
+            if (actual != null && (!actual.nodeId().equals(node.id()) || !actual.hostId().equals(node.hostId())))
+                throw new IOException("Storage node identity changed at " + url);
+            configured.add(node);
+        }
+        return configured;
+    }
+
+    private static void ensureLiveReplicasConfigured(Connection connection, List<NodeClient.Node> configured)
+        throws SQLException, IOException {
+        Set<UUID> configuredIds = new HashSet<>();
+        for (NodeClient.Node node : configured) configuredIds.add(node.id());
+        try (Statement statement = connection.createStatement();
+             ResultSet result = statement.executeQuery(
+                 "SELECT DISTINCT unnest(s.replica_ids) FROM cluster_segments s JOIN cluster_objects o ON o.generation=s.generation")) {
+            while (result.next()) {
+                UUID id = (UUID) result.getObject(1);
+                if (!configuredIds.contains(id))
+                    throw new IOException("A live segment refers to a node missing from CLUSTER_NODES: " + id);
+            }
+        }
+    }
 }

src/cloud/lunarsky/store/SigV4.java

@@ -28,28 +28,61 @@ final class SigV4 {
     }
 
     String verify(String method, URI uri, Headers headers) {
+        Map<String, String> fields = authorizationFields(headers);
+        String[] credential = credentialScope(fields.get("Credential"));
+        String date = signingDate(headers, credential[1]);
+        String payload = payloadHash(headers);
+        String signedHeaders = fields.get("SignedHeaders");
+        String canonicalHeaders = canonicalHeaders(headers, signedHeaders);
+        String canonical = method + "\n" + encode(decode(uri.getRawPath()), true) + "\n"
+            + canonicalQuery(uri.getRawQuery()) + "\n" + canonicalHeaders + "\n" + signedHeaders + "\n" + payload;
+        String scope = String.join("/", Arrays.copyOfRange(credential, 1, 5));
+        String toSign = "AWS4-HMAC-SHA256\n" + date + "\n" + scope + "\n" + hex(hash(canonical.getBytes(StandardCharsets.UTF_8)));
+        byte[] signingKey = signingKey(secretKey, credential[1], region);
+        String signature = fields.get("Signature");
+        if (!HEX.matcher(signature).matches() || !MessageDigest.isEqual(hmac(signingKey, toSign), HexFormat.of().parseHex(signature))) denied("Signature mismatch");
+        return payload;
+    }
+
+    private static Map<String, String> authorizationFields(Headers headers) {
         String authorization = single(headers, "authorization");
         if (authorization == null || !authorization.startsWith("AWS4-HMAC-SHA256 ")) denied("Signed requests are required");
-        Map<String,String> fields = new TreeMap<>();
+        Map<String, String> fields = new TreeMap<>();
         for (String part : authorization.substring(17).split(",")) {
             String[] pair = part.trim().split("=", 2);
             if (pair.length != 2 || fields.put(pair[0], pair[1]) != null) denied("Invalid authorization header");
         }
         if (!fields.keySet().equals(java.util.Set.of("Credential", "SignedHeaders", "Signature"))) denied("Invalid authorization fields");
-        String[] credential = fields.get("Credential").split("/", -1);
+        return fields;
+    }
+
+    private String[] credentialScope(String value) {
+        String[] credential = value.split("/", -1);
         if (credential.length != 5 || !credential[0].equals(accessKey) || !credential[2].equals(region)
             || !credential[3].equals("s3") || !credential[4].equals("aws4_request")) denied("Invalid credential scope");
-        String date = single(headers, "x-amz-date"), payload = single(headers, "x-amz-content-sha256");
+        return credential;
-        if (date == null || !credential[1].matches("[0-9]{8}") || !date.matches("[0-9]{8}T[0-9]{6}Z") || !date.startsWith(credential[1])) denied("Invalid signing date");
+    }
+
+    private String signingDate(Headers headers, String credentialDate) {
+        String date = single(headers, "x-amz-date");
+        if (date == null || !credentialDate.matches("[0-9]{8}") || !date.matches("[0-9]{8}T[0-9]{6}Z") || !date.startsWith(credentialDate)) denied("Invalid signing date");
         try {
             Instant signed = Instant.from(DATE.parse(date));
             if (Duration.between(signed, clock.instant()).abs().compareTo(Duration.ofMinutes(5)) > 0)
                 throw new StoreException(403, "RequestTimeTooSkewed", "Request timestamp is outside the permitted window");
         } catch (java.time.DateTimeException e) { denied("Invalid signing date"); }
+        return date;
+    }
+
+    private static String payloadHash(Headers headers) {
+        String payload = single(headers, "x-amz-content-sha256");
         if (payload == null || !HEX.matcher(payload).matches())
             throw new StoreException(400, "NotImplemented", "A hexadecimal SHA-256 payload hash is required; unsigned and chunk-signed payloads are unsupported");
         if (headers.containsKey("x-amz-security-token")) denied("Temporary credentials are unsupported");
-        String signedHeaders = fields.get("SignedHeaders");
+        return payload;
+    }
+
+    private static String canonicalHeaders(Headers headers, String signedHeaders) {
         String[] names = signedHeaders.split(";", -1);
         if (names.length > 32 || !signedHeaders.equals(String.join(";", Arrays.stream(names).distinct().sorted().toList()))) denied("Signed headers must be unique and sorted");
         var namesSet = java.util.Set.copyOf(Arrays.asList(names));
@@ -66,14 +99,7 @@ final class SigV4 {
             if (value == null) denied("Missing signed header");
             canonicalHeaders.append(name).append(':').append(value.trim().replaceAll("[\\t ]+", " ")).append('\n');
         }
-        String canonical = method + "\n" + encode(decode(uri.getRawPath()), true) + "\n"
+        return canonicalHeaders.toString();
-            + canonicalQuery(uri.getRawQuery()) + "\n" + canonicalHeaders + "\n" + signedHeaders + "\n" + payload;
-        String scope = String.join("/", Arrays.copyOfRange(credential, 1, 5));
-        String toSign = "AWS4-HMAC-SHA256\n" + date + "\n" + scope + "\n" + hex(hash(canonical.getBytes(StandardCharsets.UTF_8)));
-        byte[] signingKey = signingKey(secretKey, credential[1], region);
-        String signature = fields.get("Signature");
-        if (!HEX.matcher(signature).matches() || !MessageDigest.isEqual(hmac(signingKey, toSign), HexFormat.of().parseHex(signature))) denied("Signature mismatch");
-        return payload;
     }
 
     static String single(Headers headers, String name) {

src/cloud/lunarsky/store/Version.java

@@ -1,7 +1,7 @@
 package cloud.lunarsky.store;
 
 final class Version {
-    static final String VALUE = "0.0.2";
+    static final String VALUE = "0.0.3";
 
     private Version() {}
 }

test/cloud/lunarsky/store/HttpTest.java

@@ -66,6 +66,114 @@ public final class HttpTest {
         }
     }
 
+    private static void testObjects(HttpClient client, String base) throws Exception {
+        status(200, client.send(HttpRequest.newBuilder(URI.create(base + "/health")).GET().build(),
+            HttpResponse.BodyHandlers.ofByteArray()));
+        String key = "folder/moon-☾.txt";
+        byte[] body = "independent storage test".getBytes(StandardCharsets.UTF_8);
+        status(403, client.send(HttpRequest.newBuilder(URI.create(base + "/objects/" + key))
+            .GET().build(), HttpResponse.BodyHandlers.ofByteArray()));
+        status(200, client.send(signed(base, "PUT", key, body),
+            HttpResponse.BodyHandlers.ofByteArray()));
+        String other = "folder/stars.txt";
+        status(200, client.send(signedUri(URI.create(base + "/objects/" + other), "PUT",
+            "stars".getBytes(StandardCharsets.UTF_8), Map.of("content-type", "text/plain")),
+            HttpResponse.BodyHandlers.ofByteArray()));
+        var get = client.send(signed(base, "GET", key, new byte[0]),
+            HttpResponse.BodyHandlers.ofByteArray());
+        status(200, get);
+        if (!java.util.Arrays.equals(body, get.body())) throw new AssertionError("GET body mismatch");
+        if (!"application/octet-stream".equals(get.headers().firstValue("content-type").orElse("")))
+            throw new AssertionError("Unexpected content type");
+        var typed = client.send(signed(base, "GET", other, new byte[0]), HttpResponse.BodyHandlers.ofByteArray());
+        status(200, typed);
+        if (!"text/plain".equals(typed.headers().firstValue("content-type").orElse("")))
+            throw new AssertionError("Stored content type missing");
+        var partial = client.send(signedUri(URI.create(base + "/objects/" + other), "GET",
+            new byte[0], Map.of("range", "bytes=1-3")), HttpResponse.BodyHandlers.ofByteArray());
+        status(206, partial);
+        if (!"tar".equals(new String(partial.body(), StandardCharsets.UTF_8)) ||
+            !"bytes 1-3/5".equals(partial.headers().firstValue("content-range").orElse("")))
+            throw new AssertionError("Range response mismatch");
+        status(416, client.send(signedUri(URI.create(base + "/objects/" + other), "GET",
+            new byte[0], Map.of("range", "bytes=20-30")), HttpResponse.BodyHandlers.ofByteArray()));
+    }
+
+    private static void testListing(HttpClient client, String base) throws Exception {
+        var listed = client.send(signedUri(URI.create(base + "/objects?list-type=2&prefix=folder%2F"),
+            "GET", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofString());
+        if (listed.statusCode() != 200 || !listed.body().contains("<Key>folder/stars.txt</Key>") ||
+            !listed.body().contains("<Key>folder/moon-☾.txt</Key>"))
+            throw new AssertionError("ListObjectsV2 failed: " + listed.body());
+        var page = client.send(signedUri(URI.create(base + "/objects?list-type=2&max-keys=1"),
+            "GET", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofString());
+        if (page.statusCode() != 200 || !page.body().contains("<IsTruncated>true</IsTruncated>"))
+            throw new AssertionError("List pagination failed: " + page.body());
+        String token = page.body().split("<NextContinuationToken>")[1].split("</NextContinuationToken>")[0];
+        var next = client.send(signedUri(URI.create(base + "/objects?list-type=2&continuation-token=" + token),
+            "GET", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofString());
+        if (next.statusCode() != 200 || !next.body().contains("<Key>folder/stars.txt</Key>"))
+            throw new AssertionError("List continuation failed: " + next.body());
+        var grouped = client.send(signedUri(URI.create(base + "/objects?list-type=2&delimiter=%2F"),
+            "GET", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofString());
+        if (grouped.statusCode() != 200 || !grouped.body().contains("<CommonPrefixes><Prefix>folder/</Prefix></CommonPrefixes>") ||
+            grouped.body().contains("<Contents>"))
+            throw new AssertionError("Delimiter listing failed: " + grouped.body());
+        var encoded = client.send(signedUri(URI.create(base + "/objects?list-type=2&encoding-type=url"),
+            "GET", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofString());
+        if (encoded.statusCode() != 200 || !encoded.body().contains("folder%2Fmoon-%E2%98%BE.txt"))
+            throw new AssertionError("Encoded listing failed: " + encoded.body());
+        var emptyPage = client.send(signedUri(URI.create(base + "/objects?list-type=2&max-keys=0"),
+            "GET", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofString());
+        if (emptyPage.statusCode() != 200 || !emptyPage.body().contains("<KeyCount>0</KeyCount>"))
+            throw new AssertionError("Empty list page failed: " + emptyPage.body());
+    }
+
+    private static void testMultipart(HttpClient client, String base) throws Exception {
+        String movie = "folder/video.mp4";
+        URI initiate = URI.create(base + "/objects/" + movie + "?uploads=");
+        var created = client.send(signedUri(initiate, "POST", new byte[0],
+            Map.of("content-type", "video/mp4")), HttpResponse.BodyHandlers.ofString());
+        if (created.statusCode() != 200) throw new AssertionError("Multipart initiation failed: " + created.body());
+        String upload = created.body().split("<UploadId>")[1].split("</UploadId>")[0];
+        byte[] first = "hello ".getBytes(StandardCharsets.UTF_8);
+        byte[] second = "world".getBytes(StandardCharsets.UTF_8);
+        var partOne = client.send(signedUri(URI.create(base + "/objects/" + movie +
+            "?partNumber=1&uploadId=" + upload), "PUT", first, Map.of()), HttpResponse.BodyHandlers.ofByteArray());
+        var partTwo = client.send(signedUri(URI.create(base + "/objects/" + movie +
+            "?partNumber=2&uploadId=" + upload), "PUT", second, Map.of()), HttpResponse.BodyHandlers.ofByteArray());
+        status(200, partOne); status(200, partTwo);
+        String completion = "<CompleteMultipartUpload><Part><PartNumber>1</PartNumber><ETag>" +
+            partOne.headers().firstValue("etag").orElseThrow() +
+            "</ETag></Part><Part><PartNumber>2</PartNumber><ETag>" +
+            partTwo.headers().firstValue("etag").orElseThrow() +
+            "</ETag></Part></CompleteMultipartUpload>";
+        status(200, client.send(signedUri(URI.create(base + "/objects/" + movie + "?uploadId=" + upload),
+            "POST", completion.getBytes(StandardCharsets.UTF_8), Map.of()), HttpResponse.BodyHandlers.ofByteArray()));
+        var assembled = client.send(signed(base, "GET", movie, new byte[0]), HttpResponse.BodyHandlers.ofByteArray());
+        status(200, assembled);
+        if (!"hello world".equals(new String(assembled.body(), StandardCharsets.UTF_8)) ||
+            !"video/mp4".equals(assembled.headers().firstValue("content-type").orElse("")))
+            throw new AssertionError("Completed multipart object mismatch");
+        var abandoned = client.send(signedUri(URI.create(base + "/objects/abandoned?uploads="),
+            "POST", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofString());
+        String abandonedId = abandoned.body().split("<UploadId>")[1].split("</UploadId>")[0];
+        status(204, client.send(signedUri(URI.create(base + "/objects/abandoned?uploadId=" + abandonedId),
+            "DELETE", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofByteArray()));
+    }
+
+    private static void testDelete(HttpClient client, String base) throws Exception {
+        String key = "folder/moon-☾.txt";
+        var head = client.send(signed(base, "HEAD", key, new byte[0]),
+            HttpResponse.BodyHandlers.ofByteArray());
+        status(200, head);
+        if (head.body().length != 0) throw new AssertionError("HEAD returned a body");
+        status(204, client.send(signed(base, "DELETE", key, new byte[0]),
+            HttpResponse.BodyHandlers.ofByteArray()));
+        status(404, client.send(signed(base, "GET", key, new byte[0]),
+            HttpResponse.BodyHandlers.ofByteArray()));
+    }
+
     public static void main(String[] args) throws Exception {
         Path root = Files.createTempDirectory("store-http-test-");
         var executor = Executors.newVirtualThreadPerTaskExecutor();
@@ -79,101 +187,10 @@ public final class HttpTest {
             server.start();
             String base = "http://127.0.0.1:" + server.getAddress().getPort();
             HttpClient client = HttpClient.newHttpClient();
-            status(200, client.send(HttpRequest.newBuilder(URI.create(base + "/health")).GET().build(),
+            testObjects(client, base);
-                HttpResponse.BodyHandlers.ofByteArray()));
+            testListing(client, base);
-            String key = "folder/moon-☾.txt";
+            testMultipart(client, base);
-            byte[] body = "independent storage test".getBytes(StandardCharsets.UTF_8);
+            testDelete(client, base);
-            status(403, client.send(HttpRequest.newBuilder(URI.create(base + "/objects/" + key))
-                .GET().build(), HttpResponse.BodyHandlers.ofByteArray()));
-            status(200, client.send(signed(base, "PUT", key, body),
-                HttpResponse.BodyHandlers.ofByteArray()));
-            String other = "folder/stars.txt";
-            status(200, client.send(signedUri(URI.create(base + "/objects/" + other), "PUT",
-                "stars".getBytes(StandardCharsets.UTF_8), Map.of("content-type", "text/plain")),
-                HttpResponse.BodyHandlers.ofByteArray()));
-            var get = client.send(signed(base, "GET", key, new byte[0]),
-                HttpResponse.BodyHandlers.ofByteArray());
-            status(200, get);
-            if (!java.util.Arrays.equals(body, get.body())) throw new AssertionError("GET body mismatch");
-            if (!"application/octet-stream".equals(get.headers().firstValue("content-type").orElse("")))
-                throw new AssertionError("Unexpected content type");
-            var typed = client.send(signed(base, "GET", other, new byte[0]), HttpResponse.BodyHandlers.ofByteArray());
-            status(200, typed);
-            if (!"text/plain".equals(typed.headers().firstValue("content-type").orElse("")))
-                throw new AssertionError("Stored content type missing");
-            var partial = client.send(signedUri(URI.create(base + "/objects/" + other), "GET",
-                new byte[0], Map.of("range", "bytes=1-3")), HttpResponse.BodyHandlers.ofByteArray());
-            status(206, partial);
-            if (!"tar".equals(new String(partial.body(), StandardCharsets.UTF_8)) ||
-                !"bytes 1-3/5".equals(partial.headers().firstValue("content-range").orElse("")))
-                throw new AssertionError("Range response mismatch");
-            status(416, client.send(signedUri(URI.create(base + "/objects/" + other), "GET",
-                new byte[0], Map.of("range", "bytes=20-30")), HttpResponse.BodyHandlers.ofByteArray()));
-            var listed = client.send(signedUri(URI.create(base + "/objects?list-type=2&prefix=folder%2F"),
-                "GET", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofString());
-            if (listed.statusCode() != 200 || !listed.body().contains("<Key>folder/stars.txt</Key>") ||
-                !listed.body().contains("<Key>folder/moon-☾.txt</Key>"))
-                throw new AssertionError("ListObjectsV2 failed: " + listed.body());
-            var page = client.send(signedUri(URI.create(base + "/objects?list-type=2&max-keys=1"),
-                "GET", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofString());
-            if (page.statusCode() != 200 || !page.body().contains("<IsTruncated>true</IsTruncated>"))
-                throw new AssertionError("List pagination failed: " + page.body());
-            String token = page.body().split("<NextContinuationToken>")[1].split("</NextContinuationToken>")[0];
-            var next = client.send(signedUri(URI.create(base + "/objects?list-type=2&continuation-token=" + token),
-                "GET", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofString());
-            if (next.statusCode() != 200 || !next.body().contains("<Key>folder/stars.txt</Key>"))
-                throw new AssertionError("List continuation failed: " + next.body());
-            var grouped = client.send(signedUri(URI.create(base + "/objects?list-type=2&delimiter=%2F"),
-                "GET", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofString());
-            if (grouped.statusCode() != 200 || !grouped.body().contains("<CommonPrefixes><Prefix>folder/</Prefix></CommonPrefixes>") ||
-                grouped.body().contains("<Contents>"))
-                throw new AssertionError("Delimiter listing failed: " + grouped.body());
-            var encoded = client.send(signedUri(URI.create(base + "/objects?list-type=2&encoding-type=url"),
-                "GET", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofString());
-            if (encoded.statusCode() != 200 || !encoded.body().contains("folder%2Fmoon-%E2%98%BE.txt"))
-                throw new AssertionError("Encoded listing failed: " + encoded.body());
-            var emptyPage = client.send(signedUri(URI.create(base + "/objects?list-type=2&max-keys=0"),
-                "GET", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofString());
-            if (emptyPage.statusCode() != 200 || !emptyPage.body().contains("<KeyCount>0</KeyCount>"))
-                throw new AssertionError("Empty list page failed: " + emptyPage.body());
-            String movie = "folder/video.mp4";
-            URI initiate = URI.create(base + "/objects/" + movie + "?uploads=");
-            var created = client.send(signedUri(initiate, "POST", new byte[0],
-                Map.of("content-type", "video/mp4")), HttpResponse.BodyHandlers.ofString());
-            if (created.statusCode() != 200) throw new AssertionError("Multipart initiation failed: " + created.body());
-            String upload = created.body().split("<UploadId>")[1].split("</UploadId>")[0];
-            byte[] first = "hello ".getBytes(StandardCharsets.UTF_8);
-            byte[] second = "world".getBytes(StandardCharsets.UTF_8);
-            var partOne = client.send(signedUri(URI.create(base + "/objects/" + movie +
-                "?partNumber=1&uploadId=" + upload), "PUT", first, Map.of()), HttpResponse.BodyHandlers.ofByteArray());
-            var partTwo = client.send(signedUri(URI.create(base + "/objects/" + movie +
-                "?partNumber=2&uploadId=" + upload), "PUT", second, Map.of()), HttpResponse.BodyHandlers.ofByteArray());
-            status(200, partOne); status(200, partTwo);
-            String completion = "<CompleteMultipartUpload><Part><PartNumber>1</PartNumber><ETag>" +
-                partOne.headers().firstValue("etag").orElseThrow() +
-                "</ETag></Part><Part><PartNumber>2</PartNumber><ETag>" +
-                partTwo.headers().firstValue("etag").orElseThrow() +
-                "</ETag></Part></CompleteMultipartUpload>";
-            status(200, client.send(signedUri(URI.create(base + "/objects/" + movie + "?uploadId=" + upload),
-                "POST", completion.getBytes(StandardCharsets.UTF_8), Map.of()), HttpResponse.BodyHandlers.ofByteArray()));
-            var assembled = client.send(signed(base, "GET", movie, new byte[0]), HttpResponse.BodyHandlers.ofByteArray());
-            status(200, assembled);
-            if (!"hello world".equals(new String(assembled.body(), StandardCharsets.UTF_8)) ||
-                !"video/mp4".equals(assembled.headers().firstValue("content-type").orElse("")))
-                throw new AssertionError("Completed multipart object mismatch");
-            var abandoned = client.send(signedUri(URI.create(base + "/objects/abandoned?uploads="),
-                "POST", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofString());
-            String abandonedId = abandoned.body().split("<UploadId>")[1].split("</UploadId>")[0];
-            status(204, client.send(signedUri(URI.create(base + "/objects/abandoned?uploadId=" + abandonedId),
-                "DELETE", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofByteArray()));
-            var head = client.send(signed(base, "HEAD", key, new byte[0]),
-                HttpResponse.BodyHandlers.ofByteArray());
-            status(200, head);
-            if (head.body().length != 0) throw new AssertionError("HEAD returned a body");
-            status(204, client.send(signed(base, "DELETE", key, new byte[0]),
-                HttpResponse.BodyHandlers.ofByteArray()));
-            status(404, client.send(signed(base, "GET", key, new byte[0]),
-                HttpResponse.BodyHandlers.ofByteArray()));
             System.out.println("HTTP tests passed: health, authentication, PUT, GET, HEAD, DELETE, MIME, ranges, listing, multipart");
         } finally {
             server.stop(0);

test/cloud/lunarsky/store/StoreTest.java

@@ -14,7 +14,7 @@ public final class StoreTest {
     static ObjectStorage.Metadata put(DiskStore store,String key,byte[] body,boolean only)throws Exception{
         return store.put("test",key,new ByteArrayInputStream(body),body.length,SigV4.hex(SigV4.hash(body)),null,only,"application/octet-stream");
     }
-    public static void main(String[] args)throws Exception{
+    private static void testSignature() throws Exception {
         var headers=new com.sun.net.httpserver.Headers();
         headers.set("host","examplebucket.s3.amazonaws.com");headers.set("range","bytes=0-9");
         headers.set("x-amz-date","20130524T000000Z");
@@ -29,88 +29,116 @@ public final class StoreTest {
         fails(403,()->new SigV4("AKIAIOSFODNN7EXAMPLE","wrong","us-east-1",java.time.Clock.systemUTC()).verify("GET",uri,headers));
         headers.add("host","duplicate");fails(403,()->auth.verify("GET",uri,headers));
         System.out.println("SigV4 official vector and tampering tests passed");
-        Path root=Files.createTempDirectory("store-test-");
+    }
-        try{
+
-            try(var store=new DiskStore(root,8,10)){
+    private static void testInitialStore(Path root) throws Exception {
-                byte[] body={1,2,3,4,5,6};
+        try(var store=new DiskStore(root,8,10)){
-                put(store,"../nested/☾",body,true);
+            byte[] body={1,2,3,4,5,6};
-                try(var obj=store.open("test","../nested/☾")){if(!Arrays.equals(body,obj.stream().readAllBytes()))throw new AssertionError("Roundtrip");}
+            put(store,"../nested/☾",body,true);
-                fails(412,()->put(store,"../nested/☾",new byte[]{9},true));
+            try(var obj=store.open("test","../nested/☾")){if(!Arrays.equals(body,obj.stream().readAllBytes()))throw new AssertionError("Roundtrip");}
-                fails(507,()->put(store,"second",body,true));
+            fails(412,()->put(store,"../nested/☾",new byte[]{9},true));
-                fails(413,()->put(store,"large",new byte[9],true));
+            fails(507,()->put(store,"second",body,true));
-                fails(400,()->store.put("test","bad",new ByteArrayInputStream(body),6,"0".repeat(64),null,true,"application/octet-stream"));
+            fails(413,()->put(store,"large",new byte[9],true));
-                fails(400,()->store.put("test","short",new ByteArrayInputStream(body),7,SigV4.hex(SigV4.hash(body)),null,true,"application/octet-stream"));
+            fails(400,()->store.put("test","bad",new ByteArrayInputStream(body),6,"0".repeat(64),null,true,"application/octet-stream"));
-                fails(404,()->store.open("test","bad"));
+            fails(400,()->store.put("test","short",new ByteArrayInputStream(body),7,SigV4.hex(SigV4.hash(body)),null,true,"application/octet-stream"));
-                put(store,"../nested/☾",new byte[]{9},false);
+            fails(404,()->store.open("test","bad"));
-                put(store,"empty",new byte[0],true);
+            put(store,"../nested/☾",new byte[]{9},false);
-                if(store.list("test","","",100,null).objects().size()!=2)throw new AssertionError("List index");
+            put(store,"empty",new byte[0],true);
-                if(store.objectCount()!=2||store.legacyObjects()!=0)throw new AssertionError("Object counts");
+            if(store.list("test","","",100,null).objects().size()!=2)throw new AssertionError("List index");
-                try {
+            if(store.objectCount()!=2||store.legacyObjects()!=0)throw new AssertionError("Object counts");
-                    new DiskStore(root,8,10);
-                    throw new AssertionError("A second process opened the same data directory");
-                } catch(IOException expected) {
-                    if(!expected.getMessage().contains("already in use"))throw expected;
-                }
-            }
-            try(var restarted=new DiskStore(root,8,10)){
-                try(var obj=restarted.open("test","../nested/☾")){if(obj.stream().read()!=9)throw new AssertionError("Persistence");}
-                if(restarted.list("test","","",100,null).objects().size()!=2)throw new AssertionError("Index persistence");
-                restarted.delete("test","../nested/☾");restarted.delete("test","../nested/☾");
-                fails(404,()->restarted.open("test","../nested/☾"));
-                var uploads=new MultipartStore(restarted);
-                String upload=uploads.create("test","from-parts","text/plain");
-                byte[] part={1,2,3};
-                uploads.putPart(upload,"test","from-parts",1,new ByteArrayInputStream(part),part.length,
-                    SigV4.hex(SigV4.hash(part)),null);
-                Files.writeString(root.resolve("pending-upload-id"),upload);
-            }
-            try(var resumed=new DiskStore(root,8,10)){
-                Path unfinished=root.resolve("multipart/.creating-00000000-0000-0000-0000-000000000000");
-                Files.createDirectory(unfinished);
-                Files.write(unfinished.resolve("manifest"),new byte[]{1,2,3});
-                var uploads=new MultipartStore(resumed);
-                if(Files.exists(unfinished))throw new AssertionError("Unfinished multipart creation survived restart");
-                String upload=Files.readString(root.resolve("pending-upload-id"));
-                byte[] part={1,2,3};
-                String etag=SigV4.hex(MessageDigest.getInstance("MD5").digest(part));
-                uploads.complete(upload,"test","from-parts",List.of(new MultipartStorage.Part(1,etag)));
-                try(var obj=resumed.open("test","from-parts")){
-                    if(!Arrays.equals(part,obj.stream().readAllBytes())||!obj.metadata().contentType().equals("text/plain"))
-                        throw new AssertionError("Multipart restart");
-                }
-                resumed.delete("test","from-parts");
-            }
-            byte[] old={4,5,6};
-            String oldId=SigV4.hex(SigV4.hash("test/legacy".getBytes(java.nio.charset.StandardCharsets.UTF_8)));
-            Path oldPath=root.resolve("objects").resolve(oldId.substring(0,2)).resolve(oldId);
-            Files.createDirectories(oldPath.getParent());
-            ByteBuffer oldRecord=ByteBuffer.allocate(72+old.length).putLong(0x4c534f424a303031L)
-                .putLong(old.length).putLong(123456789L)
-                .put(MessageDigest.getInstance("MD5").digest(old)).put(SigV4.hash(old)).put(old);
-            Files.write(oldPath,oldRecord.array());
-            try(var migrated=new DiskStore(root,8,10)){
-                if(migrated.objectCount()!=2||migrated.legacyObjects()!=1)throw new AssertionError("Legacy counts");
-                try(var obj=migrated.open("test","legacy")){
-                    if(!Arrays.equals(old,obj.stream().readAllBytes()))throw new AssertionError("Legacy read");
-                }
-                if(migrated.list("test","","",100,null).objects().stream().anyMatch(entry->entry.key().equals("legacy")))
-                    throw new AssertionError("Legacy object appeared without a stored key");
-                put(migrated,"legacy",old,false);
-                if(migrated.objectCount()!=2||migrated.legacyObjects()!=0)throw new AssertionError("Legacy count after overwrite");
-                if(migrated.list("test","","",100,null).objects().stream().noneMatch(entry->entry.key().equals("legacy")))
-                    throw new AssertionError("Legacy overwrite was not indexed");
-            }
-            Files.delete(root.resolve("pending-upload-id"));
-            String id=SigV4.hex(SigV4.hash("test/empty".getBytes(java.nio.charset.StandardCharsets.UTF_8)));
-            Files.write(root.resolve("objects").resolve(id.substring(0,2)).resolve(id),new byte[]{1},StandardOpenOption.APPEND);
             try {
                 new DiskStore(root,8,10);
-                throw new AssertionError("A damaged object record was accepted");
+                throw new AssertionError("A second process opened the same data directory");
             } catch(IOException expected) {
-                if(!expected.getMessage().contains("object record"))throw expected;
+                if(!expected.getMessage().contains("already in use"))throw expected;
             }
-            try(var pending=Files.list(root.resolve("pending"))){if(pending.count()!=0)throw new AssertionError("Pending cleanup");}
+        }
+    }
+
+    private static void testRestart(Path root) throws Exception {
+        try(var restarted=new DiskStore(root,8,10)){
+            try(var obj=restarted.open("test","../nested/☾")){if(obj.stream().read()!=9)throw new AssertionError("Persistence");}
+            if(restarted.list("test","","",100,null).objects().size()!=2)throw new AssertionError("Index persistence");
+            restarted.delete("test","../nested/☾");restarted.delete("test","../nested/☾");
+            fails(404,()->restarted.open("test","../nested/☾"));
+            var uploads=new MultipartStore(restarted);
+            String upload=uploads.create("test","from-parts","text/plain");
+            byte[] part={1,2,3};
+            uploads.putPart(upload,"test","from-parts",1,new ByteArrayInputStream(part),part.length,
+                SigV4.hex(SigV4.hash(part)),null);
+            Files.writeString(root.resolve("pending-upload-id"),upload);
+        }
+    }
+
+    private static void testMultipartRecovery(Path root) throws Exception {
+        try(var resumed=new DiskStore(root,8,10)){
+            Path unfinished=root.resolve("multipart/.creating-00000000-0000-0000-0000-000000000000");
+            Files.createDirectory(unfinished);
+            Files.write(unfinished.resolve("manifest"),new byte[]{1,2,3});
+            var uploads=new MultipartStore(resumed);
+            if(Files.exists(unfinished))throw new AssertionError("Unfinished multipart creation survived restart");
+            String upload=Files.readString(root.resolve("pending-upload-id"));
+            byte[] part={1,2,3};
+            String etag=SigV4.hex(MessageDigest.getInstance("MD5").digest(part));
+            uploads.complete(upload,"test","from-parts",List.of(new MultipartStorage.Part(1,etag)));
+            try(var obj=resumed.open("test","from-parts")){
+                if(!Arrays.equals(part,obj.stream().readAllBytes())||!obj.metadata().contentType().equals("text/plain"))
+                    throw new AssertionError("Multipart restart");
+            }
+            resumed.delete("test","from-parts");
+        }
+    }
+
+    private static void testLegacyRecord(Path root) throws Exception {
+        byte[] old={4,5,6};
+        String oldId=SigV4.hex(SigV4.hash("test/legacy".getBytes(java.nio.charset.StandardCharsets.UTF_8)));
+        Path oldPath=root.resolve("objects").resolve(oldId.substring(0,2)).resolve(oldId);
+        Files.createDirectories(oldPath.getParent());
+        ByteBuffer oldRecord=ByteBuffer.allocate(72+old.length).putLong(0x4c534f424a303031L)
+            .putLong(old.length).putLong(123456789L)
+            .put(MessageDigest.getInstance("MD5").digest(old)).put(SigV4.hash(old)).put(old);
+        Files.write(oldPath,oldRecord.array());
+        try(var migrated=new DiskStore(root,8,10)){
+            if(migrated.objectCount()!=2||migrated.legacyObjects()!=1)throw new AssertionError("Legacy counts");
+            try(var obj=migrated.open("test","legacy")){
+                if(!Arrays.equals(old,obj.stream().readAllBytes()))throw new AssertionError("Legacy read");
+            }
+            if(migrated.list("test","","",100,null).objects().stream().anyMatch(entry->entry.key().equals("legacy")))
+                throw new AssertionError("Legacy object appeared without a stored key");
+            put(migrated,"legacy",old,false);
+            if(migrated.objectCount()!=2||migrated.legacyObjects()!=0)throw new AssertionError("Legacy count after overwrite");
+            if(migrated.list("test","","",100,null).objects().stream().noneMatch(entry->entry.key().equals("legacy")))
+                throw new AssertionError("Legacy overwrite was not indexed");
+        }
+    }
+
+    private static void testCorruption(Path root) throws Exception {
+        Files.delete(root.resolve("pending-upload-id"));
+        String id=SigV4.hex(SigV4.hash("test/empty".getBytes(java.nio.charset.StandardCharsets.UTF_8)));
+        Files.write(root.resolve("objects").resolve(id.substring(0,2)).resolve(id),new byte[]{1},StandardOpenOption.APPEND);
+        try {
+            new DiskStore(root,8,10);
+            throw new AssertionError("A damaged object record was accepted");
+        } catch(IOException expected) {
+            if(!expected.getMessage().contains("object record"))throw expected;
+        }
+        try(var pending=Files.list(root.resolve("pending"))){if(pending.count()!=0)throw new AssertionError("Pending cleanup");}
+    }
+
+    public static void main(String[] args) throws Exception {
+        testSignature();
+        Path root = Files.createTempDirectory("store-test-");
+        try {
+            testInitialStore(root);
+            testRestart(root);
+            testMultipartRecovery(root);
+            testLegacyRecord(root);
+            testCorruption(root);
             System.out.println("Java storage tests passed: roundtrip, quota, indexing, persistence, multipart recovery, legacy reads, locking, corruption, delete");
-        }finally{try(var paths=Files.walk(root)){for(var p:paths.sorted(java.util.Comparator.reverseOrder()).toList())Files.delete(p);}}
+        } finally {
+            try (var paths = Files.walk(root)) {
+                for (var path : paths.sorted(java.util.Comparator.reverseOrder()).toList()) Files.delete(path);
+            }
+        }
     }
 }