Release ObjectStore 0.0.4

README.md

@@ -20,6 +20,7 @@ Source: [GitHub](https://github.com/LunarSkyOSS/ObjectStore) · [Gitea mirror](h
 - [Local cluster prototype](#local-cluster-prototype)
 - [Migrating a local cluster](#migrating-a-local-cluster)
 - [Adding a cluster node](#adding-a-cluster-node)
+- [Cluster maintenance and recovery](#cluster-maintenance-and-recovery)
 - [Limits and safety](#limits-and-safety)
 - [Disclaimer](#disclaimer)
 - [AI contributions](#ai-contributions)
@@ -32,7 +33,8 @@ ObjectStore serves one configured bucket.
 - ✅ Configurable per-object and total logical size limits
 - ✅ CLI status, version, and full payload verification
 - ✅ Local cluster prototype with stable node IDs and host-aware placement code
-- ⬜ Automatic repair, rebalance, and garbage collection
+- ✅ Opt-in automatic repair, rebalance, and guarded garbage collection in the local cluster
+- ✅ Metadata backup and tested restore to a separate local PostgreSQL instance
 - ⬜ Production multi-server deployment and metadata failover
 
 New objects retain their content type and key. Objects written by the earlier single-node format remain readable, but cannot appear in listings until overwritten because their original keys were not stored.
@@ -43,14 +45,16 @@ New objects retain their content type and key. Objects written by the earlier si
 - ✅ `PutObject`, `GetObject`, `HeadObject`, and `DeleteObject` in both modes
 - ✅ Single-range GET and `ListObjectsV2` in both modes
 - ✅ SHA-256 payload verification and `x-amz-checksum-sha256` in both modes
-- ✅ `CreateMultipartUpload`, `UploadPart`, `CompleteMultipartUpload`, and `AbortMultipartUpload` in single-node mode
-- ⬜ Multipart uploads in cluster mode
+- ✅ `Content-MD5` and CRC32, CRC32C, SHA-1, SHA-256, SHA-512, and MD5 checksum headers on `PutObject` and `UploadPart`
+- ✅ `CreateMultipartUpload`, `UploadPart`, `CompleteMultipartUpload`, and `AbortMultipartUpload` in both modes
+- ✅ `ListParts` and `ListMultipartUploads` in both modes
 - ⬜ Presigned URLs and streaming Signature V4 uploads
-- ⬜ `CopyObject`, `ListParts`, and `ListMultipartUploads`
-- ⬜ `Content-MD5` and checksum algorithms other than SHA-256
+- ✅ `CopyObject` within the configured bucket, with `COPY` and `REPLACE` content-type behavior
+- ⬜ CRC64NVME and XXHash checksums, checksum trailers, and persisted non-SHA-256 checksum metadata
 - ⬜ Bucket creation and listing, object versioning, ACLs, tags, and user metadata
 
 This is an S3 API subset, not full AWS S3 compatibility. Unsupported S3 operations and Amazon-specific headers are rejected.
+Checksum values are validated before an object or part is published. Non-SHA-256 checksums are returned on upload but are not stored for later reads. Copies use the existing object size limit and do not support cross-bucket or versioned sources.
 
 ## Single-node setup
 
@@ -92,6 +96,8 @@ docker compose --env-file /path/to/cluster.env -f compose.cluster.yaml run --rm
 
 The cluster S3 endpoint binds to `127.0.0.1:9001`; storage nodes and PostgreSQL have no published ports. The separate repair container holds the repair credential and restores missing or corrupt replicas.
 
+Multipart parts are stored on cluster nodes and indexed in PostgreSQL. Incomplete uploads count toward the logical capacity limit; abort them to release that capacity. Repair includes staged parts. The gateway upgrades the metadata schema when it starts, so back up the database before upgrading an existing cluster.
+
 Node UUIDs persist on their volumes, and replica manifests use those UUIDs so reordering configured URLs cannot move an existing replica. Each node also has an operator-assigned physical host UUID. New writes require acknowledgements from two different host UUIDs. The optional `CLUSTER_TEST_NODE_DOMAINS=true` override counts containers instead, solely for local process tests; all containers in this Compose file share one physical host.
 
 ## Migrating a local cluster
@@ -108,11 +114,23 @@ The old format did not record the original URL mapping, so the inventory check i
 
 ## Adding a cluster node
 
-`objectstore cluster-join http://new-node:9100 expected-host-uuid` registers an additional local node. Add its URL to `CLUSTER_NODES` and restart the gateway to use it for new writes. Existing segments stay where their manifests say; this is capacity expansion for new writes, not a rebalance. `scripts/test-cluster.sh` exercises a fourth container joining and receiving new segments.
+`objectstore cluster-join http://new-node:9100 expected-host-uuid` registers an additional local node. Add its URL to `CLUSTER_NODES` and restart the gateway. A repair pass then copies existing segments to their preferred nodes and removes obsolete replicas from the manifest only after verifying the replacements. `scripts/test-cluster.sh` exercises a fourth container joining, receiving new segments, and rebalancing existing ones.
+
+## Cluster maintenance and recovery
+
+From the Compose directory, use `docker compose --env-file /path/to/cluster.env -f compose.cluster.yaml` as the command prefix:
+
+```sh
+docker compose --env-file /path/to/cluster.env -f compose.cluster.yaml --profile automatic up -d maintenance
+docker compose --env-file /path/to/cluster.env -f compose.cluster.yaml run --rm gc
+sh scripts/backup-cluster-metadata.sh /path/to/cluster.env /path/to/metadata.dump
+```
+
+The maintenance service is opt-in. It repairs missing or corrupt replicas and rebalances them every 60 seconds by default. Set `CLUSTER_MAINTENANCE_INTERVAL_SECONDS` to change the interval. The `gc` command is a dry run; use `gc --apply` only after checking its candidate count and keeping independent backups. Cleanup records each orphan on one pass and waits at least `CLUSTER_GC_MIN_AGE_SECONDS` before deleting it on a later pass. The default age is 14 days. To permit deletion, set `CLUSTER_BACKUP_RETENTION_SECONDS` to your actual backup retention in seconds; it must be at least one day and shorter than the cleanup age. Scheduled cleanup also requires `CLUSTER_GC_ENABLED=true` on the maintenance service. The backup command writes a verified PostgreSQL archive with private file permissions. Restore it to a separate database and point a gateway at that database only after validating the restore. A backup restore is manual recovery, not automatic failover.
 
 ## Limits and safety
 
-The cluster retains old and failed-write segments. It has no garbage collection, metadata standby, automated rebalance, private-network TLS, scoped credentials, or physical host verification yet. Host UUIDs are operator labels, not proof that machines have separate power, disks, or network paths. Keep `CLUSTER_LOCAL_DEV=true` limited to local tests.
+The local cluster has no automatic metadata failover, private-network TLS, scoped credentials, or physical host verification. Garbage collection can remove data required by an older metadata backup, so its retention guard is essential. Host UUIDs are operator labels, not proof that machines have separate power, disks, or network paths. Keep `CLUSTER_LOCAL_DEV=true` limited to local tests.
 
 The standalone cluster node binds to localhost by default. Set `NODE_BIND` only for a private test network; the Compose file binds inside its private Docker network. PostgreSQL JDBC 42.7.14 is bundled in the image with its license inside the JAR.
 

TESTS.md

@@ -18,8 +18,8 @@ The script compiles the source and test programs into `out/classes`, then runs:
 | --- | --- |
 | `StoreTest` | Signature V4 test vector and tampering, local writes and reads, quotas, restart persistence, multipart recovery, legacy reads, locking, and corruption rejection. |
 | `ConcurrencyTest` | Atomic local overwrites and consistent reads, listings, and deletes during concurrent access. |
-| `HttpTest` | Signed HTTP requests, object operations, ranges, listing, and single-node multipart uploads. |
-| `ClusterNodeTest` | Node identity and locking, authenticated segment transfers, checksum rejection, repair authorization, and restart cleanup. |
+| `HttpTest` | Signed HTTP requests, object operations, same-bucket copies, checksum acceptance and rejection, ranges, listing, and multipart uploads in single-node mode. |
+| `ClusterNodeTest` | Node identity and locking, authenticated segment transfers, checksum rejection, repair authorization, inventory and guarded deletion, and restart cleanup. |
 | `CliTest` | Version, status, verification, and a nonzero result for corrupt data. |
 
 The script exits nonzero on failure. The test programs use temporary local directories and loopback HTTP ports; they do not use an existing ObjectStore volume.
@@ -42,12 +42,12 @@ COMPOSE_PROJECT_NAME=objectstore-tests sh scripts/test-cluster.sh /tmp/objectsto
 Use a fresh, disposable Compose project. The script writes test objects, stops and restarts storage nodes and PostgreSQL, corrupts a replica to exercise repair, and joins a fourth node. It leaves the test stack running. To remove **only that test project's** containers and volumes after review:
 
 ```sh
-COMPOSE_PROJECT_NAME=objectstore-tests docker compose --env-file /tmp/objectstore-cluster-tests.env -f compose.cluster.yaml --profile expansion down -v
+COMPOSE_PROJECT_NAME=objectstore-tests docker compose --env-file /tmp/objectstore-cluster-tests.env -f compose.cluster.yaml --profile expansion --profile automatic --profile recovery down -v
 ```
 
 If port 9001 is occupied, set `CLUSTER_HOST_PORT` to the same free port in both the environment file and the shell before running the script. The script reads that port from the shell; Compose reads it from the file.
 
-The Docker suite checks signed S3 operations, multi-segment objects, concurrent overwrites, reads and writes with a node stopped, refusal to write without a storage quorum, restart recovery, corrupt-replica repair, metadata unavailability, and placement on a newly joined node. It also checks that containers labeled as one physical host cannot satisfy the normal host quorum. Its local-only override permits the remaining phases to use containers as separate test domains.
+The Docker suite checks signed S3 operations, same-bucket copies, upload checksums, multi-segment objects, concurrent overwrites, multipart staging and listings, completion after a gateway restart and node loss, reads and writes with a node stopped, refusal to write without a storage quorum, restart recovery, corrupt-replica repair including staged parts, metadata unavailability, and placement on a newly joined node. It then checks rebalance to the fourth node, automatic repair, garbage collection dry run and delayed deletion, and a metadata backup restored to a separate PostgreSQL instance while the primary is stopped. It also checks that containers labeled as one physical host cannot satisfy the normal host quorum. Its local-only override permits the remaining phases to use containers as separate test domains.
 
 `ClusterMigrationTest` is a separate legacy-format fixture and is **not** run by either test script. Do not run its `create` phase against a populated metadata database. The migration procedure is in the [README](README.md#migrating-a-local-cluster).
 
@@ -55,6 +55,6 @@ The Docker suite checks signed S3 operations, multi-segment objects, concurrent
 
 - Container stops are not physical power cuts or disk failures. The automated suite does not reboot a host or test every possible crash point.
 - The Compose nodes share one machine. Passing the local-only quorum override does not demonstrate durability across independent hosts, racks, or sites.
-- The suite does not test metadata failover, an off-site backup restore, prolonged load, or full AWS S3 compatibility.
+- The suite does not test automatic metadata failover, an off-site backup restore, prolonged load, or full AWS S3 compatibility.
 
 See [Limits and safety](README.md#limits-and-safety) before evaluating any multi-server deployment.

compose.cluster.yaml

@@ -59,7 +59,25 @@ services:
     security_opt:
       - no-new-privileges:true
 
-  repair:
+  metadata-recovery:
+    image: postgres:17-alpine
+    profiles: [recovery]
+    environment:
+      POSTGRES_DB: objectstore
+      POSTGRES_USER: objectstore
+      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:?Set POSTGRES_PASSWORD}
+    volumes:
+      - cluster-metadata-recovery:/var/lib/postgresql/data
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U objectstore -d objectstore"]
+      interval: 5s
+      timeout: 3s
+      retries: 10
+    mem_limit: 256m
+    security_opt:
+      - no-new-privileges:true
+
+  repair: &repair
     image: lunarsky-objectstore:cluster-local
     profiles: [maintenance]
     entrypoint: ["/usr/local/bin/objectstore", "cluster-repair"]
@@ -74,6 +92,50 @@ services:
       POSTGRES_JDBC_URL: jdbc:postgresql://metadata:5432/objectstore?connectTimeout=3&socketTimeout=10
       POSTGRES_USER: objectstore
       POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:?Set POSTGRES_PASSWORD}
+      CLUSTER_GC_MIN_AGE_SECONDS: ${CLUSTER_GC_MIN_AGE_SECONDS:-1209600}
+      CLUSTER_BACKUP_RETENTION_SECONDS: ${CLUSTER_BACKUP_RETENTION_SECONDS:-0}
+      CLUSTER_GC_TEST_MODE: ${CLUSTER_GC_TEST_MODE:-false}
+    mem_limit: 384m
+    security_opt:
+      - no-new-privileges:true
+    cap_drop:
+      - ALL
+
+  gc:
+    <<: *repair
+    entrypoint: ["/usr/local/bin/objectstore", "cluster-gc"]
+
+  maintenance:
+    image: lunarsky-objectstore:cluster-local
+    profiles: [automatic]
+    restart: unless-stopped
+    entrypoint: ["/usr/local/bin/objectstore", "cluster-repair", "--loop"]
+    environment:
+      STORE_MODE: cluster
+      CLUSTER_LOCAL_DEV: "true"
+      CLUSTER_TEST_NODE_DOMAINS: "true"
+      CLUSTER_NODES: ${CLUSTER_NODES:-http://node-a:9100,http://node-b:9100,http://node-c:9100}
+      CLUSTER_TOKEN: ${CLUSTER_TOKEN:?Set CLUSTER_TOKEN}
+      CLUSTER_REPAIR_TOKEN: ${CLUSTER_REPAIR_TOKEN:?Set CLUSTER_REPAIR_TOKEN}
+      CLUSTER_MAINTENANCE_INTERVAL_SECONDS: ${CLUSTER_MAINTENANCE_INTERVAL_SECONDS:-60}
+      CLUSTER_GC_ENABLED: ${CLUSTER_GC_ENABLED:-false}
+      CLUSTER_GC_INTERVAL_SECONDS: ${CLUSTER_GC_INTERVAL_SECONDS:-86400}
+      CLUSTER_GC_MIN_AGE_SECONDS: ${CLUSTER_GC_MIN_AGE_SECONDS:-1209600}
+      CLUSTER_BACKUP_RETENTION_SECONDS: ${CLUSTER_BACKUP_RETENTION_SECONDS:-0}
+      CLUSTER_GC_TEST_MODE: ${CLUSTER_GC_TEST_MODE:-false}
+      S3_BUCKET: ${S3_BUCKET:-objects}
+      POSTGRES_JDBC_URL: jdbc:postgresql://metadata:5432/objectstore?connectTimeout=3&socketTimeout=10
+      POSTGRES_USER: objectstore
+      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:?Set POSTGRES_PASSWORD}
+    depends_on:
+      metadata:
+        condition: service_healthy
+      node-a:
+        condition: service_healthy
+      node-b:
+        condition: service_healthy
+      node-c:
+        condition: service_healthy
     mem_limit: 384m
     security_opt:
       - no-new-privileges:true
@@ -121,6 +183,7 @@ services:
 
 volumes:
   cluster-metadata:
+  cluster-metadata-recovery:
   cluster-node-a:
   cluster-node-b:
   cluster-node-c:

scripts/backup-cluster-metadata.sh

@@ -0,0 +1,19 @@
+#!/bin/sh
+set -eu
+cd "$(dirname "$0")/.."
+env_file=${1:?Usage: sh scripts/backup-cluster-metadata.sh /path/to/cluster.env /path/to/metadata.dump}
+output=${2:?Usage: sh scripts/backup-cluster-metadata.sh /path/to/cluster.env /path/to/metadata.dump}
+if [ -e "$output" ]; then
+  echo "Backup destination already exists" >&2
+  exit 1
+fi
+umask 077
+temporary=$(mktemp "${output}.tmp.XXXXXX")
+trap 'rm -f "$temporary"' EXIT
+docker compose --env-file "$env_file" -f compose.cluster.yaml exec -T metadata \
+  pg_dump -U objectstore -d objectstore --format=custom --no-owner --no-acl > "$temporary"
+test -s "$temporary"
+docker compose --env-file "$env_file" -f compose.cluster.yaml exec -T metadata \
+  pg_restore --list < "$temporary" > /dev/null
+mv "$temporary" "$output"
+echo "Metadata backup written to $output"

scripts/objectstore

@@ -3,6 +3,10 @@ if [ "${1:-}" = "cluster-repair" ]; then
   shift
   exec java -XX:MaxRAMPercentage=70 --add-modules java.net.http -cp /app:/app/postgresql.jar cloud.lunarsky.store.ClusterRepair "$@"
 fi
+if [ "${1:-}" = "cluster-gc" ]; then
+  shift
+  exec java -XX:MaxRAMPercentage=70 --add-modules java.net.http -cp /app:/app/postgresql.jar cloud.lunarsky.store.ClusterGc "$@"
+fi
 if [ "${1:-}" = "cluster-migrate" ]; then
   shift
   exec java -XX:MaxRAMPercentage=70 --add-modules java.net.http -cp /app:/app/postgresql.jar cloud.lunarsky.store.ClusterMigrate "$@"

scripts/test-cluster-http.py

@@ -1,11 +1,14 @@
 #!/usr/bin/env python3
 import datetime
+import base64
 import hashlib
 import hmac
 import http.client
 import pathlib
+import re
 import sys
 import urllib.parse
+import zlib
 
 
 values = dict(line.strip().split("=", 1) for line in pathlib.Path(sys.argv[1]).read_text().splitlines()
@@ -43,7 +46,7 @@ def request(method, path, body=b"", extra=None):
                                 f"SignedHeaders={signed_names},Signature={signature}")
     connection = http.client.HTTPConnection("127.0.0.1", port, timeout=30)
     try:
-        connection.request(method, path, body=body if method == "PUT" else None, headers=headers)
+        connection.request(method, path, body=body if method in ("PUT", "POST") else None, headers=headers)
         response = connection.getresponse()
         return response.status, response.read(), response.headers
     finally:
@@ -81,4 +84,58 @@ status, _, _ = request("DELETE", key)
 assert status == 204, status
 status, _, _ = request("GET", key)
 assert status == 404, status
-print("Cluster HTTP tests passed: signed PUT, GET, range, HEAD, LIST, DELETE")
+
+copy_source = f"/{bucket}/cluster-test/copy-source.txt"
+copy_target = f"/{bucket}/cluster-test/copied.txt"
+body = b"cluster copy and checksum test"
+crc32 = base64.b64encode(zlib.crc32(body).to_bytes(4, "big")).decode()
+md5 = base64.b64encode(hashlib.md5(body).digest()).decode()
+status, _, headers = request("PUT", copy_source, body,
+                             {"content-type": "text/plain", "content-md5": md5,
+                              "x-amz-checksum-crc32": crc32,
+                              "x-amz-sdk-checksum-algorithm": "CRC32"})
+assert status == 200 and headers["x-amz-checksum-crc32"] == crc32, status
+status, content, _ = request("PUT", copy_source, body,
+                             {"content-md5": base64.b64encode(bytes(16)).decode()})
+assert status == 400 and b"BadDigest" in content, (status, content)
+status, content, _ = request("GET", copy_source)
+assert status == 200 and content == body, (status, content)
+status, content, _ = request("PUT", copy_target, extra={"x-amz-copy-source": copy_source})
+assert status == 200 and b"<CopyObjectResult>" in content, (status, content)
+status, content, headers = request("GET", copy_target)
+assert status == 200 and content == body and headers["content-type"] == "text/plain", (status, content)
+status, _, _ = request("DELETE", copy_source)
+assert status == 204, status
+status, _, _ = request("DELETE", copy_target)
+assert status == 204, status
+
+multipart_key = f"/{bucket}/cluster-test/http-multipart.txt"
+status, content, _ = request("POST", multipart_key + "?uploads")
+assert status == 200, (status, content)
+match = re.search(rb"<UploadId>([0-9a-fA-F]{8}(?:-[0-9a-fA-F]{4}){3}-[0-9a-fA-F]{12})</UploadId>",
+                  content[:8192])
+assert match, content
+upload_id = match.group(1).decode("ascii")
+part_etags = []
+for number, part in enumerate((b"hello ", b"world"), start=1):
+    checksum = base64.b64encode(hashlib.sha1(part).digest()).decode()
+    status, _, headers = request("PUT", multipart_key + f"?partNumber={number}&uploadId={upload_id}",
+                                 part, {"x-amz-checksum-sha1": checksum})
+    assert status == 200, status
+    assert headers["x-amz-checksum-sha1"] == checksum, headers
+    part_etags.append(headers["etag"])
+status, content, _ = request("GET", multipart_key + f"?uploadId={upload_id}&max-parts=1")
+assert status == 200 and b"<IsTruncated>true</IsTruncated>" in content, (status, content)
+status, content, _ = request("GET", f"/{bucket}?uploads&prefix=cluster-test%2Fhttp-multipart")
+assert status == 200 and upload_id.encode() in content, (status, content)
+completion = "<CompleteMultipartUpload>" + "".join(
+    f"<Part><PartNumber>{number}</PartNumber><ETag>{etag}</ETag></Part>"
+    for number, etag in enumerate(part_etags, start=1)) + "</CompleteMultipartUpload>"
+status, content, _ = request("POST", multipart_key + f"?uploadId={upload_id}", completion.encode(),
+                             {"content-type": "application/xml"})
+assert status == 200 and b"<CompleteMultipartUploadResult>" in content, (status, content)
+status, content, _ = request("GET", multipart_key)
+assert status == 200 and content == b"hello world", (status, content)
+status, content, _ = request("GET", f"/{bucket}?uploads&prefix=cluster-test%2Fhttp-multipart")
+assert status == 200 and upload_id.encode() not in content, (status, content)
+print("Cluster HTTP tests passed: signed objects, copies, checksums, and multipart operations")

scripts/test-cluster.sh

@@ -3,8 +3,13 @@ set -eu
 cd "$(dirname "$0")/.."
 env_file=${1:?Usage: sh scripts/test-cluster.sh /path/to/local-cluster.env}
 host_port=${CLUSTER_HOST_PORT:-9001}
+backup_dir=
 compose() { docker compose --env-file "$env_file" -f compose.cluster.yaml "$@"; }
-restore() { compose start metadata node-a node-b >/dev/null 2>&1 || true; }
+restore() {
+  compose stop maintenance >/dev/null 2>&1 || true
+  compose start metadata node-a node-b node-c >/dev/null 2>&1 || true
+  if [ -n "$backup_dir" ]; then rm -rf "$backup_dir"; fi
+}
 trap restore EXIT
 compose up -d --build
 run_phase() {
@@ -20,10 +25,24 @@ wait_ready() {
   done
 }
 run_phase basic
+run_phase multipart-stage
+compose restart gateway
+wait_ready
+part_segment=$(compose exec -T metadata psql -U objectstore -d objectstore -At -c \
+  "SELECT segment_id FROM cluster_upload_segments LIMIT 1")
+printf '%s\n' "$part_segment" | grep -Eq '^[0-9a-f-]{36}$'
+part_shard=$(printf '%s' "$part_segment" | cut -c1-2)
+compose exec -T node-a sh -c 'printf corrupted > "/data/segments/$1/$2"' _ "$part_shard" "$part_segment"
+compose run --rm -T repair
+part_expected=$(compose exec -T metadata psql -U objectstore -d objectstore -At -c \
+  "SELECT encode(sha256,'hex') FROM cluster_upload_segments WHERE segment_id='$part_segment'")
+part_actual=$(compose exec -T node-a sha256sum "/data/segments/$part_shard/$part_segment" | cut -d' ' -f1)
+[ "$part_expected" = "$part_actual" ]
 run_phase same-host
 run_phase concurrent
 compose stop node-a
 run_phase degraded
+run_phase multipart-complete
 compose stop node-b
 run_phase quorum-lost
 compose start node-a node-b
@@ -55,4 +74,58 @@ compose up -d --no-deps gateway
 wait_ready
 run_phase recovered
 run_phase joined
+compose run --rm -T repair
+run_phase balanced
+export CLUSTER_MAINTENANCE_INTERVAL_SECONDS=1
+compose --profile automatic up -d maintenance
+node_a_id=$(compose exec -T metadata psql -U objectstore -d objectstore -At -c \
+  "SELECT node_id FROM cluster_nodes WHERE endpoint='http://node-a:9100'")
+segment_id=$(compose exec -T metadata psql -U objectstore -d objectstore -At -c \
+  "SELECT s.segment_id FROM cluster_segments s JOIN cluster_objects o ON o.generation=s.generation WHERE '$node_a_id'::uuid = ANY(s.replica_ids) LIMIT 1")
+expected=$(compose exec -T metadata psql -U objectstore -d objectstore -At -c \
+  "SELECT encode(s.sha256,'hex') FROM cluster_segments s WHERE s.segment_id='$segment_id' LIMIT 1")
+shard=$(printf '%s' "$segment_id" | cut -c1-2)
+compose exec -T node-a sh -c 'printf corrupted > "/data/segments/$1/$2"' _ "$shard" "$segment_id"
+attempt=0
+while :; do
+  actual=$(compose exec -T node-a sha256sum "/data/segments/$shard/$segment_id" | cut -d' ' -f1)
+  [ "$actual" = "$expected" ] && break
+  attempt=$((attempt + 1))
+  [ "$attempt" -lt 90 ] || { echo 'Automatic repair did not restore the replica' >&2; exit 1; }
+  sleep 1
+done
+compose stop maintenance
+export CLUSTER_GC_TEST_MODE=true CLUSTER_GC_MIN_AGE_SECONDS=0
+compose stop node-c
+if gc_refusal=$(compose run --rm -T gc --apply 2>&1); then
+  echo 'Cleanup proceeded while replicas needed repair' >&2
+  exit 1
+fi
+printf '%s\n' "$gc_refusal" | grep -q 'Refusing cleanup while live segments need repair'
+compose start node-c
+before=$(compose exec -T metadata psql -U objectstore -d objectstore -At -c \
+  'SELECT count(*) FROM cluster_gc_candidates')
+compose run --rm -T gc
+after=$(compose exec -T metadata psql -U objectstore -d objectstore -At -c \
+  'SELECT count(*) FROM cluster_gc_candidates')
+[ "$before" = "$after" ]
+first_gc=$(compose run --rm -T gc --apply)
+printf '%s\n' "$first_gc" | grep -q '^orphan_candidates=[1-9]'
+printf '%s\n' "$first_gc" | grep -q '^segments_deleted=0$'
+second_gc=$(compose run --rm -T gc --apply)
+printf '%s\n' "$second_gc" | grep -q '^segments_deleted=[1-9]'
+run_phase recovered
+run_phase verify-expanded
+backup_dir=$(mktemp -d)
+sh scripts/backup-cluster-metadata.sh "$env_file" "$backup_dir/metadata.dump"
+compose --profile recovery up -d --wait metadata-recovery
+compose exec -T metadata-recovery pg_restore -U objectstore -d objectstore --no-owner --no-acl \
+  < "$backup_dir/metadata.dump"
+compose stop metadata
+compose run --rm -T --no-deps \
+  -e 'POSTGRES_JDBC_URL=jdbc:postgresql://metadata-recovery:5432/objectstore?connectTimeout=3&socketTimeout=10' \
+  --entrypoint java gateway --add-modules jdk.httpserver,java.net.http \
+  -cp /app:/app/postgresql.jar cloud.lunarsky.store.ClusterIntegrationTest recovered
+compose start metadata
+wait_ready
 echo 'Cluster failure tests passed'

src/cloud/lunarsky/store/Cli.java

@@ -109,7 +109,9 @@ public final class Cli {
             }
             if (verify) {
                 MessageDigest sha = digest("SHA-256"), md5 = digest("MD5");
-                byte[] buffer = new byte[65536]; long count = 0; int n;
+                byte[] buffer = new byte[65536];
+                long count = 0;
+                int n;
                 while ((n = input.read(buffer)) != -1) {
                     count += n;
                     sha.update(buffer, 0, n);

src/cloud/lunarsky/store/ClusterGc.java

@@ -0,0 +1,43 @@
+package cloud.lunarsky.store;
+
+import java.net.URI;
+import java.util.Arrays;
+import java.util.Map;
+
+public final class ClusterGc {
+    public static void main(String[] args) throws Exception {
+        if (args.length > 1 || (args.length == 1 && !args[0].equals("--apply")))
+            throw new IllegalArgumentException("Usage: objectstore cluster-gc [--apply]");
+        run(System.getenv(), args.length == 1);
+    }
+
+    static void run(Map<String, String> env, boolean apply) throws Exception {
+        if (!"cluster".equals(env.get("STORE_MODE")) || !"true".equals(env.get("CLUSTER_LOCAL_DEV")))
+            throw new IllegalArgumentException("Cluster garbage collection is only enabled in local cluster mode");
+        boolean testDomains = "true".equals(env.get("CLUSTER_TEST_NODE_DOMAINS"));
+        boolean disposableTest = testDomains && "true".equals(env.get("CLUSTER_GC_TEST_MODE"));
+        long age = Long.parseLong(env.getOrDefault("CLUSTER_GC_MIN_AGE_SECONDS", "1209600"));
+        long backupRetention = Long.parseLong(env.getOrDefault("CLUSTER_BACKUP_RETENTION_SECONDS", "0"));
+        if (age < 0 || age > 315360000 || (age == 0 && !disposableTest))
+            throw new IllegalArgumentException("Invalid garbage collection age");
+        if (apply && !disposableTest && (backupRetention < 86400 || age <= backupRetention))
+            throw new IllegalArgumentException("Set a garbage collection age longer than the backup retention");
+        try (ClusterStore store = new ClusterStore(env.get("POSTGRES_JDBC_URL"), env.get("POSTGRES_USER"),
+                env.get("POSTGRES_PASSWORD"), env.get("S3_BUCKET"),
+                Arrays.stream(env.get("CLUSTER_NODES").split(",")).map(URI::create).toList(),
+                env.get("CLUSTER_TOKEN"), env.get("CLUSTER_REPAIR_TOKEN"), 134217728, 2147483648L,
+                testDomains)) {
+            if (apply) {
+                var repair = store.repairOnce();
+                if (repair.underReplicated() > 0 || repair.unrecoverable() > 0)
+                    throw new IllegalStateException("Refusing cleanup while live segments need repair");
+            }
+            var report = store.collectGarbage(age * 1000, apply);
+            System.out.println("segments_scanned=" + report.scanned());
+            System.out.println("orphan_candidates=" + report.eligible());
+            System.out.println("segments_deleted=" + report.deleted());
+            System.out.println("unavailable_nodes=" + report.unavailableNodes());
+            if (report.unavailableNodes() > 0) throw new IllegalStateException("Cleanup did not scan every node");
+        }
+    }
+}

src/cloud/lunarsky/store/ClusterMigrate.java

@@ -148,7 +148,8 @@ public final class ClusterMigrate {
             }
             connection.commit();
         } catch (SQLException | IOException | RuntimeException error) {
-            try { connection.rollback(); } catch (SQLException rollback) { error.addSuppressed(rollback); }
+            try { connection.rollback(); }
+            catch (SQLException rollback) { error.addSuppressed(rollback); }
             if (error instanceof IOException io) throw io;
             if (error instanceof SQLException sql) throw sql;
             throw (RuntimeException) error;

src/cloud/lunarsky/store/ClusterNode.java

@@ -15,7 +15,9 @@ import java.nio.file.StandardCopyOption;
 import java.nio.file.StandardOpenOption;
 import java.security.MessageDigest;
 import java.util.Arrays;
+import java.util.Comparator;
 import java.util.HexFormat;
+import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.Executors;
@@ -84,11 +86,18 @@ public final class ClusterNode implements AutoCloseable {
                 return;
             }
             if (!expectedNodeAndRepairAuthorized(exchange)) return;
+            if (path.equals("/segments") && exchange.getRequestMethod().equals("GET")) {
+                if (maintenanceAuthorized(exchange)) inventory(exchange);
+                return;
+            }
             String id = segmentId(exchange, path);
             if (id == null) return;
             switch (exchange.getRequestMethod()) {
                 case "PUT" -> put(exchange, segmentPath(id, true));
                 case "GET" -> get(exchange, segmentPath(id, false));
+                case "DELETE" -> {
+                    if (maintenanceAuthorized(exchange)) delete(exchange, segmentPath(id, false));
+                }
                 default -> respond(exchange, 405, "Method not allowed");
             }
         } catch (IllegalArgumentException error) {
@@ -130,6 +139,77 @@ public final class ClusterNode implements AutoCloseable {
         return true;
     }
 
+    private boolean maintenanceAuthorized(HttpExchange exchange) throws IOException {
+        String supplied = exchange.getRequestHeaders().getFirst("X-Cluster-Repair-Token");
+        byte[] value = supplied == null ? new byte[0] : supplied.getBytes(java.nio.charset.StandardCharsets.UTF_8);
+        if (MessageDigest.isEqual(repairToken, value)) return true;
+        respond(exchange, 403, "Repair authority required");
+        return false;
+    }
+
+    private void inventory(HttpExchange exchange) throws IOException {
+        String query = exchange.getRequestURI().getRawQuery();
+        if (query == null || !query.matches("shard=[0-9a-f]{2}(&after=[0-9a-f-]{36})?")) {
+            respond(exchange, 400, "Invalid inventory request");
+            return;
+        }
+        String shard = query.substring(6, 8);
+        String after = query.length() > 8 ? query.substring(15) : "";
+        if (!after.isEmpty() && (!after.startsWith(shard) || !UUID.fromString(after).toString().equals(after))) {
+            respond(exchange, 400, "Invalid inventory cursor");
+            return;
+        }
+        Path directory = segments.resolve(shard);
+        if (!Files.isDirectory(directory)) {
+            respond(exchange, 200, "");
+            return;
+        }
+        List<Path> files;
+        try (var entries = Files.list(directory)) {
+            files = entries.filter(Files::isRegularFile).sorted(Comparator.comparing(path ->
+                path.getFileName().toString())).toList();
+        }
+        StringBuilder body = new StringBuilder();
+        int count = 0;
+        for (Path file : files) {
+            String id = file.getFileName().toString();
+            if (id.compareTo(after) <= 0 || !id.matches("[0-9a-f-]{36}")) continue;
+            if (!UUID.fromString(id).toString().equals(id)) continue;
+            body.append(id).append(' ').append(Files.getLastModifiedTime(file).toMillis()).append('\n');
+            if (++count == 1000) break;
+        }
+        respond(exchange, 200, body.toString());
+    }
+
+    private synchronized void delete(HttpExchange exchange, Path target) throws IOException {
+        String expected = exchange.getRequestHeaders().getFirst("X-Cluster-Expected-Mtime");
+        String age = exchange.getRequestHeaders().getFirst("X-Cluster-Gc-Min-Age-Millis");
+        long expectedTime, minimumAge;
+        try {
+            expectedTime = Long.parseLong(expected);
+            minimumAge = Long.parseLong(age);
+        } catch (NumberFormatException error) {
+            respond(exchange, 400, "Invalid deletion guard");
+            return;
+        }
+        if (minimumAge < 0 || minimumAge > System.currentTimeMillis()) {
+            respond(exchange, 400, "Invalid deletion age");
+            return;
+        }
+        if (!Files.isRegularFile(target)) {
+            exchange.sendResponseHeaders(404, -1);
+            return;
+        }
+        long modified = Files.getLastModifiedTime(target).toMillis();
+        if (modified != expectedTime || modified > System.currentTimeMillis() - minimumAge) {
+            exchange.sendResponseHeaders(409, -1);
+            return;
+        }
+        Files.delete(target);
+        DiskStore.syncDirectory(target.getParent());
+        exchange.sendResponseHeaders(204, -1);
+    }
+
     private static String segmentId(HttpExchange exchange, String path) throws IOException {
         if (!path.matches("/segments/[0-9a-f-]{36}")) {
             respond(exchange, 404, "Not found");
@@ -252,7 +332,8 @@ public final class ClusterNode implements AutoCloseable {
         Runtime.getRuntime().addShutdownHook(new Thread(() -> {
             server.stop(5);
             executor.close();
-            try { node.close(); } catch (IOException error) { System.err.println("Node close failed: " + error); }
+            try { node.close(); }
+            catch (IOException error) { System.err.println("Node close failed: " + error); }
         }));
         server.start();
         System.out.println("ObjectStore cluster node listening on :" + port);

src/cloud/lunarsky/store/ClusterRepair.java

@@ -6,10 +6,20 @@ import java.util.Map;
 
 public final class ClusterRepair {
     public static void main(String[] args) throws Exception {
-        if (args.length != 0) throw new IllegalArgumentException("Usage: objectstore cluster-repair");
+        if (args.length > 1 || (args.length == 1 && !args[0].equals("--loop")))
+            throw new IllegalArgumentException("Usage: objectstore cluster-repair [--loop]");
         Map<String, String> env = System.getenv();
         if (!"cluster".equals(env.get("STORE_MODE")) || !"true".equals(env.get("CLUSTER_LOCAL_DEV")))
             throw new IllegalArgumentException("Cluster repair is only enabled in local cluster mode");
+        boolean loop = args.length == 1;
+        long seconds = Long.parseLong(env.getOrDefault("CLUSTER_MAINTENANCE_INTERVAL_SECONDS", "60"));
+        if (seconds < 1 || seconds > 3600) throw new IllegalArgumentException("Invalid maintenance interval");
+        boolean gcEnabled = loop && "true".equals(env.get("CLUSTER_GC_ENABLED"));
+        long gcInterval = Long.parseLong(env.getOrDefault("CLUSTER_GC_INTERVAL_SECONDS", "86400"));
+        if (gcEnabled && (gcInterval < 1 || gcInterval > 604800))
+            throw new IllegalArgumentException("Invalid garbage collection interval");
+        long nextGc = 0;
+        do {
             try (ClusterStore store = new ClusterStore(env.get("POSTGRES_JDBC_URL"), env.get("POSTGRES_USER"),
                     env.get("POSTGRES_PASSWORD"), env.get("S3_BUCKET"),
                     Arrays.stream(env.get("CLUSTER_NODES").split(",")).map(URI::create).toList(),
@@ -18,9 +28,19 @@ public final class ClusterRepair {
                 var report = store.repairOnce();
                 System.out.println("segments_scanned=" + report.scanned());
                 System.out.println("replicas_restored=" + report.restored());
+                System.out.println("segments_rebalanced=" + report.rebalanced());
                 System.out.println("segments_under_replicated=" + report.underReplicated());
                 System.out.println("segments_unrecoverable=" + report.unrecoverable());
-            if (report.unrecoverable() > 0) System.exit(1);
+                if (!loop && report.unrecoverable() > 0) System.exit(1);
+                if (gcEnabled && System.currentTimeMillis() >= nextGc) {
+                    ClusterGc.run(env, true);
+                    nextGc = System.currentTimeMillis() + gcInterval * 1000;
                 }
+            } catch (Exception error) {
+                if (!loop) throw error;
+                System.err.println("Cluster maintenance failed: " + error.getMessage());
+            }
+            if (loop) Thread.sleep(seconds * 1000);
+        } while (loop);
     }
 }

src/cloud/lunarsky/store/ClusterStore.java

@@ -18,16 +18,21 @@ import java.sql.SQLException;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Base64;
+import java.util.Comparator;
 import java.util.HexFormat;
 import java.util.HashSet;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Set;
 import java.util.UUID;
 
-final class ClusterStore implements ObjectStorage {
+final class ClusterStore implements ObjectStorage, MultipartStorage {
     private record Segment(UUID id, int length, byte[] hash, List<UUID> replicas) {}
-    private record RepairTarget(UUID generation, int ordinal, long version, Segment segment) {}
-    record RepairReport(int scanned, int restored, int underReplicated, int unrecoverable) {}
+    private record RepairTarget(UUID id, int part, int ordinal, long version, Segment segment) {}
+    private record Upload(String contentType) {}
+    private record StoredPart(long length, String etag, List<Segment> segments) {}
+    record RepairReport(int scanned, int restored, int rebalanced, int underReplicated, int unrecoverable) {}
+    record GcReport(int scanned, int eligible, int deleted, int unavailableNodes) {}
     private final String jdbcUrl, user, password, configuredBucket;
     private final NodeClient nodes;
     private final long maxObject, maxTotal;
@@ -38,8 +43,12 @@ final class ClusterStore implements ObjectStorage {
                  boolean testNodeDomains) throws IOException {
         if (jdbcUrl == null || !jdbcUrl.startsWith("jdbc:postgresql://") || user == null || password == null)
             throw new IllegalArgumentException("Invalid metadata database configuration");
-        this.jdbcUrl = jdbcUrl; this.user = user; this.password = password;
-        this.configuredBucket = bucket; this.maxObject = maxObject; this.maxTotal = maxTotal;
+        this.jdbcUrl = jdbcUrl;
+        this.user = user;
+        this.password = password;
+        this.configuredBucket = bucket;
+        this.maxObject = maxObject;
+        this.maxTotal = maxTotal;
         this.testNodeDomains = testNodeDomains;
         try (Connection connection = connect()) {
             int format = SchemaMigrator.prepare(connection, bucket);
@@ -50,22 +59,29 @@ final class ClusterStore implements ObjectStorage {
 
     private Connection connect() throws SQLException { return DriverManager.getConnection(jdbcUrl, user, password); }
 
+    private static void lockGc(Connection connection, boolean shared) throws SQLException {
+        try (var statement = connection.createStatement()) {
+            statement.execute("SELECT pg_advisory_lock" + (shared ? "_shared" : "") + "(6834071092783)");
+        }
+    }
+
     @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);
+            byte[] fullHash = stageInput(staged, input, length, expectedHash, checksum, md5);
             checkCapacity(bucket, key, length, createOnly);
-            segments = uploadSegments(staged, length);
-        } finally { Files.deleteIfExists(staged); }
+            try (Connection connection = connect()) {
+                lockGc(connection, true);
+                List<Segment> segments = uploadSegments(staged, length);
                 Metadata metadata = new Metadata(length, Instant.now().toEpochMilli(),
                     HexFormat.of().formatHex(md5.digest()), fullHash, bucket, key, contentType);
-        persistObject(metadata, segments, createOnly);
+                persistObject(connection, metadata, segments, createOnly);
                 return metadata;
+            } catch (SQLException error) { throw databaseError(error); }
+        } finally { Files.deleteIfExists(staged); }
     }
 
     private void validatePut(String bucket, long length, String contentType) {
@@ -88,7 +104,8 @@ final class ClusterStore implements ObjectStorage {
                 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);
+                sha.update(buffer, 0, count);
+                md5.update(buffer, 0, count);
                 output.write(buffer, 0, count);
                 remaining -= count;
             }
@@ -111,7 +128,8 @@ final class ClusterStore implements ObjectStorage {
                 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)
+                    if (maxTotal - (result.getLong(1) - Math.max(0, previous)) -
+                        stagedBytes(connection, bucket) < length)
                         throw new StoreException(507, "InsufficientStorage", "Store capacity limit reached");
                 }
             }
@@ -152,25 +170,30 @@ final class ClusterStore implements ObjectStorage {
         return segments;
     }
 
-    private void persistObject(Metadata metadata, List<Segment> segments, boolean createOnly) throws IOException {
+    private void persistObject(Connection connection, 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()) {
+        try {
             connection.setAutoCommit(false);
             try {
                 long used = lockUsage(connection, bucket);
                 long previous = currentLength(connection, bucket, key);
                 if (createOnly && previous >= 0)
                     throw new StoreException(412, "PreconditionFailed", "Object already exists");
-                if (used - Math.max(0, previous) > maxTotal - length)
+                if (maxTotal - (used - Math.max(0, previous)) -
+                    stagedBytes(connection, bucket) < length)
                     throw new StoreException(507, "InsufficientStorage", "Store capacity limit reached");
                 try (PreparedStatement insert = connection.prepareStatement(
                     "INSERT INTO cluster_segments (generation, ordinal, segment_id, length, sha256, replicas, replica_ids) VALUES (?, ?, ?, ?, ?, 'v2', ?)")) {
                     for (int i = 0; i < segments.size(); i++) {
                         Segment segment = segments.get(i);
-                        insert.setObject(1, generation); insert.setInt(2, i); insert.setObject(3, segment.id());
-                        insert.setInt(4, segment.length()); insert.setBytes(5, segment.hash());
+                        insert.setObject(1, generation);
+                        insert.setInt(2, i);
+                        insert.setObject(3, segment.id());
+                        insert.setInt(4, segment.length());
+                        insert.setBytes(5, segment.hash());
                         insert.setArray(6, connection.createArrayOf("uuid", segment.replicas().toArray()));
                         insert.addBatch();
                     }
@@ -178,13 +201,18 @@ final class ClusterStore implements ObjectStorage {
                 }
                 try (PreparedStatement update = connection.prepareStatement(
                     "INSERT INTO cluster_objects VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (bucket, object_key) DO UPDATE SET generation=EXCLUDED.generation, length=EXCLUDED.length, modified=EXCLUDED.modified, etag=EXCLUDED.etag, sha256=EXCLUDED.sha256, content_type=EXCLUDED.content_type")) {
-                    bindObject(update, metadata, generation); update.executeUpdate();
+                    bindObject(update, metadata, generation);
+                    update.executeUpdate();
                 }
                 try (PreparedStatement update = connection.prepareStatement("UPDATE cluster_usage SET used_bytes=? WHERE bucket=?")) {
-                    update.setLong(1, used - Math.max(0, previous) + length); update.setString(2, bucket); update.executeUpdate();
+                    update.setLong(1, used - Math.max(0, previous) + length);
+                    update.setString(2, bucket);
+                    update.executeUpdate();
                 }
                 try (PreparedStatement delete = connection.prepareStatement("DELETE FROM cluster_tombstones WHERE bucket=? AND object_key=?")) {
-                    delete.setString(1, bucket); delete.setString(2, key); delete.executeUpdate();
+                    delete.setString(1, bucket);
+                    delete.setString(2, key);
+                    delete.executeUpdate();
                 }
                 connection.commit();
             } catch (SQLException | RuntimeException error) {
@@ -195,16 +223,288 @@ final class ClusterStore implements ObjectStorage {
         } catch (SQLException error) { throw databaseError(error); }
     }
 
+    @Override public String create(String bucket, String key, String contentType) throws IOException {
+        if (!configuredBucket.equals(bucket)) throw new StoreException(404, "NoSuchBucket", "Bucket not found");
+        if (contentType == null || contentType.getBytes(java.nio.charset.StandardCharsets.UTF_8).length > 255)
+            throw new StoreException(400, "InvalidArgument", "Invalid Content-Type");
+        UUID id = UUID.randomUUID();
+        try (Connection connection = connect()) {
+            connection.setAutoCommit(false);
+            try {
+                lockUsage(connection, bucket);
+                try (PreparedStatement count = connection.prepareStatement("SELECT count(*) FROM cluster_uploads WHERE bucket=?")) {
+                    count.setString(1, bucket);
+                    try (ResultSet result = count.executeQuery()) {
+                        result.next();
+                        if (result.getLong(1) >= 32)
+                            throw new StoreException(503, "SlowDown", "Too many active uploads");
+                    }
+                }
+                try (PreparedStatement insert = connection.prepareStatement(
+                    "INSERT INTO cluster_uploads VALUES (?, ?, ?, ?, ?)")) {
+                    insert.setObject(1, id);
+                    insert.setString(2, bucket);
+                    insert.setString(3, key);
+                    insert.setString(4, contentType);
+                    insert.setLong(5, Instant.now().toEpochMilli());
+                    insert.executeUpdate();
+                }
+                connection.commit();
+            } catch (SQLException | RuntimeException error) {
+                connection.rollback();
+                if (error instanceof SQLException sql) throw databaseError(sql);
+                throw error;
+            }
+        } catch (SQLException error) { throw databaseError(error); }
+        return id.toString();
+    }
+
+    @Override public String putPart(String id, String bucket, String key, int number, InputStream input,
+                                    long length, String expectedHash, String checksum) throws IOException {
+        if (number < 1 || number > 10000) throw new StoreException(400, "InvalidArgument", "Invalid part number");
+        validatePut(bucket, length, "application/octet-stream");
+        UUID uploadId = uploadId(id);
+        try (Connection connection = connect()) { upload(connection, uploadId, bucket, key, false); }
+        catch (SQLException error) { throw databaseError(error); }
+        Path staged = Files.createTempFile("objectstore-part-", ".pending");
+        MessageDigest md5 = digest("MD5");
+        try {
+            stageInput(staged, input, length, expectedHash, checksum, md5);
+            try (Connection connection = connect()) {
+                lockGc(connection, true);
+                List<Segment> segments = uploadSegments(staged, length);
+                String etag = HexFormat.of().formatHex(md5.digest());
+                connection.setAutoCommit(false);
+                try {
+                    long used = lockUsage(connection, bucket);
+                    upload(connection, uploadId, bucket, key, true);
+                    long previous = partLength(connection, uploadId, number);
+                    if (maxTotal - used - (stagedBytes(connection, bucket) - previous) < length)
+                        throw new StoreException(507, "InsufficientStorage", "Multipart staging limit reached");
+                    try (PreparedStatement insert = connection.prepareStatement(
+                        "INSERT INTO cluster_upload_parts VALUES (?, ?, ?, ?, ?) ON CONFLICT (upload_id, part_number) DO UPDATE SET length=EXCLUDED.length, etag=EXCLUDED.etag, modified=EXCLUDED.modified")) {
+                        insert.setObject(1, uploadId);
+                        insert.setInt(2, number);
+                        insert.setLong(3, length);
+                        insert.setString(4, etag);
+                        insert.setLong(5, Instant.now().toEpochMilli());
+                        insert.executeUpdate();
+                    }
+                    try (PreparedStatement delete = connection.prepareStatement(
+                        "DELETE FROM cluster_upload_segments WHERE upload_id=? AND part_number=?")) {
+                        delete.setObject(1, uploadId);
+                        delete.setInt(2, number);
+                        delete.executeUpdate();
+                    }
+                    try (PreparedStatement insert = connection.prepareStatement(
+                        "INSERT INTO cluster_upload_segments VALUES (?, ?, ?, ?, ?, ?, ?)")) {
+                        for (int ordinal = 0; ordinal < segments.size(); ordinal++) {
+                            Segment segment = segments.get(ordinal);
+                            insert.setObject(1, uploadId);
+                            insert.setInt(2, number);
+                            insert.setInt(3, ordinal);
+                            insert.setObject(4, segment.id());
+                            insert.setInt(5, segment.length());
+                            insert.setBytes(6, segment.hash());
+                            insert.setArray(7, connection.createArrayOf("uuid", segment.replicas().toArray()));
+                            insert.addBatch();
+                        }
+                        insert.executeBatch();
+                    }
+                    connection.commit();
+                } catch (SQLException | RuntimeException error) {
+                    connection.rollback();
+                    if (error instanceof SQLException sql) throw databaseError(sql);
+                    throw error;
+                }
+                return etag;
+            } catch (SQLException error) { throw databaseError(error); }
+        } finally { Files.deleteIfExists(staged); }
+    }
+
+    @Override public Metadata complete(String id, String bucket, String key, List<MultipartStorage.Part> parts) throws IOException {
+        if (parts.isEmpty() || parts.size() > 10000)
+            throw new StoreException(400, "InvalidPart", "No valid parts supplied");
+        UUID uploadId = uploadId(id);
+        try (Connection connection = connect()) {
+            connection.setAutoCommit(false);
+            try {
+                long used = lockUsage(connection, bucket);
+                Upload upload = upload(connection, uploadId, bucket, key, true);
+                long staged = stagedBytes(connection, bucket);
+                long uploadBytes = uploadLength(connection, uploadId);
+                MessageDigest fullHash = digest("SHA-256");
+                MessageDigest etagHash = digest("MD5");
+                List<Segment> selected = new ArrayList<>();
+                long total = 0;
+                int last = 0;
+                for (MultipartStorage.Part requested : parts) {
+                    if (requested.number() <= last || requested.number() > 10000)
+                        throw new StoreException(400, "InvalidPartOrder", "Parts must be in ascending order");
+                    last = requested.number();
+                    StoredPart part = storedPart(connection, uploadId, requested.number());
+                    if (part == null || !part.etag().equals(requested.etag().replace("\"", "")))
+                        throw new StoreException(400, "InvalidPart", "Part ETag mismatch");
+                    if (part.length() > maxObject - total)
+                        throw new StoreException(413, "EntityTooLarge", "Object exceeds the configured size limit");
+                    total += part.length();
+                    MessageDigest partHash = digest("MD5");
+                    for (Segment segment : part.segments()) {
+                        byte[] bytes = readableSegment(segment);
+                        if (bytes == null) throw new StoreException(503, "SlowDown", "A part has no verified replica");
+                        fullHash.update(bytes);
+                        partHash.update(bytes);
+                        selected.add(segment);
+                    }
+                    byte[] md5 = partHash.digest();
+                    if (!part.etag().equals(HexFormat.of().formatHex(md5)))
+                        throw new StoreException(503, "SlowDown", "A part failed integrity verification");
+                    etagHash.update(md5);
+                }
+                long previous = currentLength(connection, bucket, key);
+                if (maxTotal - (used - Math.max(0, previous)) - (staged - uploadBytes) < total)
+                    throw new StoreException(507, "InsufficientStorage", "Store capacity limit reached");
+                Metadata metadata = new Metadata(total, Instant.now().toEpochMilli(),
+                    HexFormat.of().formatHex(etagHash.digest()) + "-" + parts.size(), fullHash.digest(),
+                    bucket, key, upload.contentType());
+                UUID generation = UUID.randomUUID();
+                try (PreparedStatement insert = connection.prepareStatement(
+                    "INSERT INTO cluster_segments (generation, ordinal, segment_id, length, sha256, replicas, replica_ids) VALUES (?, ?, ?, ?, ?, 'v2', ?)")) {
+                    for (int ordinal = 0; ordinal < selected.size(); ordinal++) {
+                        Segment segment = selected.get(ordinal);
+                        insert.setObject(1, generation);
+                        insert.setInt(2, ordinal);
+                        insert.setObject(3, segment.id());
+                        insert.setInt(4, segment.length());
+                        insert.setBytes(5, segment.hash());
+                        insert.setArray(6, connection.createArrayOf("uuid", segment.replicas().toArray()));
+                        insert.addBatch();
+                    }
+                    insert.executeBatch();
+                }
+                try (PreparedStatement update = connection.prepareStatement(
+                    "INSERT INTO cluster_objects VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (bucket, object_key) DO UPDATE SET generation=EXCLUDED.generation, length=EXCLUDED.length, modified=EXCLUDED.modified, etag=EXCLUDED.etag, sha256=EXCLUDED.sha256, content_type=EXCLUDED.content_type")) {
+                    bindObject(update, metadata, generation);
+                    update.executeUpdate();
+                }
+                try (PreparedStatement update = connection.prepareStatement("UPDATE cluster_usage SET used_bytes=? WHERE bucket=?")) {
+                    update.setLong(1, used - Math.max(0, previous) + total);
+                    update.setString(2, bucket);
+                    update.executeUpdate();
+                }
+                try (PreparedStatement delete = connection.prepareStatement("DELETE FROM cluster_tombstones WHERE bucket=? AND object_key=?")) {
+                    delete.setString(1, bucket);
+                    delete.setString(2, key);
+                    delete.executeUpdate();
+                }
+                try (PreparedStatement delete = connection.prepareStatement("DELETE FROM cluster_uploads WHERE upload_id=?")) {
+                    delete.setObject(1, uploadId);
+                    delete.executeUpdate();
+                }
+                connection.commit();
+                return metadata;
+            } catch (SQLException | IOException | RuntimeException error) {
+                connection.rollback();
+                if (error instanceof SQLException sql) throw databaseError(sql);
+                if (error instanceof IOException io) throw io;
+                throw error;
+            }
+        } catch (SQLException error) { throw databaseError(error); }
+    }
+
+    @Override public void abort(String id, String bucket, String key) throws IOException {
+        UUID uploadId = uploadId(id);
+        try (Connection connection = connect()) {
+            connection.setAutoCommit(false);
+            try {
+                lockUsage(connection, bucket);
+                upload(connection, uploadId, bucket, key, true);
+                try (PreparedStatement delete = connection.prepareStatement("DELETE FROM cluster_uploads WHERE upload_id=?")) {
+                    delete.setObject(1, uploadId);
+                    delete.executeUpdate();
+                }
+                connection.commit();
+            } catch (SQLException | RuntimeException error) {
+                connection.rollback();
+                if (error instanceof SQLException sql) throw databaseError(sql);
+                throw error;
+            }
+        } catch (SQLException error) { throw databaseError(error); }
+    }
+
+    @Override public PartPage listParts(String id, String bucket, String key, int marker, int maxParts)
+            throws IOException {
+        UUID uploadId = uploadId(id);
+        try (Connection connection = connect()) {
+            upload(connection, uploadId, bucket, key, false);
+            List<PartInfo> parts = new ArrayList<>();
+            boolean truncated = false;
+            try (PreparedStatement query = connection.prepareStatement(
+                "SELECT part_number, length, etag, modified FROM cluster_upload_parts WHERE upload_id=? AND part_number>? ORDER BY part_number LIMIT ?")) {
+                query.setObject(1, uploadId);
+                query.setInt(2, marker);
+                query.setInt(3, maxParts + 1);
+                try (ResultSet result = query.executeQuery()) {
+                    while (result.next()) {
+                        if (parts.size() == maxParts) {
+                            truncated = true;
+                            break;
+                        }
+                        parts.add(new PartInfo(result.getInt(1), result.getLong(2), result.getString(3), result.getLong(4)));
+                    }
+                }
+            }
+            int next = parts.isEmpty() ? marker : parts.getLast().number();
+            return new PartPage(parts, next, truncated);
+        } catch (SQLException error) { throw databaseError(error); }
+    }
+
+    @Override public List<UploadInfo> listUploads(String bucket, String prefix) throws IOException {
+        if (!configuredBucket.equals(bucket)) throw new StoreException(404, "NoSuchBucket", "Bucket not found");
+        List<UploadInfo> uploads = new ArrayList<>();
+        try (Connection connection = connect(); PreparedStatement query = connection.prepareStatement(
+            "SELECT upload_id, object_key, created_at FROM cluster_uploads WHERE bucket=?")) {
+            query.setString(1, bucket);
+            try (ResultSet result = query.executeQuery()) {
+                while (result.next()) {
+                    String key = result.getString(2);
+                    if (key.startsWith(prefix))
+                        uploads.add(new UploadInfo(result.getObject(1).toString(), key, result.getLong(3)));
+                }
+            }
+        } catch (SQLException error) { throw databaseError(error); }
+        uploads.sort(Comparator.comparing(UploadInfo::key).thenComparing(UploadInfo::id));
+        return uploads;
+    }
+
+    @Override public int activeUploads() {
+        try (Connection connection = connect(); PreparedStatement query = connection.prepareStatement(
+            "SELECT count(*) FROM cluster_uploads WHERE bucket=?")) {
+            query.setString(1, configuredBucket);
+            try (ResultSet result = query.executeQuery()) {
+                result.next();
+                return result.getInt(1);
+            }
+        } catch (SQLException error) { throw new IllegalStateException("Could not count multipart uploads", error); }
+    }
+
+    @Override public long stagedBytes() {
+        try (Connection connection = connect()) { return stagedBytes(connection, configuredBucket); }
+        catch (SQLException error) { throw new IllegalStateException("Could not count staged bytes", error); }
+    }
+
     @Override public OpenObject open(String bucket, String key) throws IOException {
         try (Connection connection = connect()) {
             connection.setAutoCommit(false);
             connection.setTransactionIsolation(Connection.TRANSACTION_REPEATABLE_READ);
+            lockGc(connection, true);
             try {
                 Metadata metadata;
                 UUID generation;
                 try (PreparedStatement query = connection.prepareStatement(
                     "SELECT generation, length, modified, etag, sha256, content_type FROM cluster_objects WHERE bucket=? AND object_key=?")) {
-                    query.setString(1, bucket); query.setString(2, key);
+                    query.setString(1, bucket);
+                    query.setString(2, key);
                     try (ResultSet result = query.executeQuery()) {
                         if (!result.next()) throw new StoreException(404, "NoSuchKey", "Object not found");
                         generation = (UUID) result.getObject(1);
@@ -245,17 +545,23 @@ final class ClusterStore implements ObjectStorage {
                 long used = lockUsage(connection, bucket);
                 long previous = currentLength(connection, bucket, key);
                 try (PreparedStatement delete = connection.prepareStatement("DELETE FROM cluster_objects WHERE bucket=? AND object_key=?")) {
-                    delete.setString(1, bucket); delete.setString(2, key); delete.executeUpdate();
+                    delete.setString(1, bucket);
+                    delete.setString(2, key);
+                    delete.executeUpdate();
                 }
                 try (PreparedStatement update = connection.prepareStatement(
                     "INSERT INTO cluster_tombstones VALUES (?, ?, ?, ?) ON CONFLICT (bucket, object_key) DO UPDATE SET generation=EXCLUDED.generation, deleted_at=EXCLUDED.deleted_at")) {
-                    update.setString(1, bucket); update.setString(2, key);
-                    update.setObject(3, UUID.randomUUID()); update.setLong(4, Instant.now().toEpochMilli());
+                    update.setString(1, bucket);
+                    update.setString(2, key);
+                    update.setObject(3, UUID.randomUUID());
+                    update.setLong(4, Instant.now().toEpochMilli());
                     update.executeUpdate();
                 }
                 if (previous >= 0) {
                     try (PreparedStatement update = connection.prepareStatement("UPDATE cluster_usage SET used_bytes=? WHERE bucket=?")) {
-                        update.setLong(1, used - previous); update.setString(2, bucket); update.executeUpdate();
+                        update.setLong(1, used - previous);
+                        update.setString(2, bucket);
+                        update.executeUpdate();
                     }
                 }
                 connection.commit();
@@ -297,10 +603,18 @@ final class ClusterStore implements ObjectStorage {
             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 {
+            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;
@@ -325,16 +639,122 @@ final class ClusterStore implements ObjectStorage {
             }
         }
     }
+
+    private static UUID uploadId(String id) {
+        try {
+            if (id == null || !id.matches("[0-9a-f-]{36}")) throw new IllegalArgumentException();
+            return UUID.fromString(id);
+        } catch (IllegalArgumentException error) {
+            throw new StoreException(404, "NoSuchUpload", "Upload not found");
+        }
+    }
+
+    private static Upload upload(Connection connection, UUID id, String bucket, String key, boolean lock)
+            throws SQLException {
+        String sql = "SELECT bucket, object_key, content_type, created_at FROM cluster_uploads WHERE upload_id=?" +
+            (lock ? " FOR UPDATE" : "");
+        try (PreparedStatement query = connection.prepareStatement(sql)) {
+            query.setObject(1, id);
+            try (ResultSet result = query.executeQuery()) {
+                if (!result.next() || !result.getString(1).equals(bucket) || !result.getString(2).equals(key))
+                    throw new StoreException(404, "NoSuchUpload", "Upload not found");
+                return new Upload(result.getString(3));
+            }
+        }
+    }
+
+    private static long stagedBytes(Connection connection, String bucket) throws SQLException {
+        try (PreparedStatement query = connection.prepareStatement(
+            "SELECT COALESCE(sum(p.length), 0) FROM cluster_upload_parts p JOIN cluster_uploads u USING (upload_id) WHERE u.bucket=?")) {
+            query.setString(1, bucket);
+            try (ResultSet result = query.executeQuery()) {
+                result.next();
+                return result.getLong(1);
+            }
+        }
+    }
+
+    private static long uploadLength(Connection connection, UUID id) throws SQLException {
+        try (PreparedStatement query = connection.prepareStatement(
+            "SELECT COALESCE(sum(length), 0) FROM cluster_upload_parts WHERE upload_id=?")) {
+            query.setObject(1, id);
+            try (ResultSet result = query.executeQuery()) {
+                result.next();
+                return result.getLong(1);
+            }
+        }
+    }
+
+    private static long partLength(Connection connection, UUID id, int number) throws SQLException {
+        try (PreparedStatement query = connection.prepareStatement(
+            "SELECT length FROM cluster_upload_parts WHERE upload_id=? AND part_number=?")) {
+            query.setObject(1, id);
+            query.setInt(2, number);
+            try (ResultSet result = query.executeQuery()) {
+                return result.next() ? result.getLong(1) : 0;
+            }
+        }
+    }
+
+    private static StoredPart storedPart(Connection connection, UUID id, int number) throws SQLException, IOException {
+        long length;
+        String etag;
+        try (PreparedStatement query = connection.prepareStatement(
+            "SELECT length, etag FROM cluster_upload_parts WHERE upload_id=? AND part_number=?")) {
+            query.setObject(1, id);
+            query.setInt(2, number);
+            try (ResultSet result = query.executeQuery()) {
+                if (!result.next()) return null;
+                length = result.getLong(1);
+                etag = result.getString(2);
+            }
+        }
+        List<Segment> segments = new ArrayList<>();
+        try (PreparedStatement query = connection.prepareStatement(
+            "SELECT ordinal, segment_id, length, sha256, replica_ids FROM cluster_upload_segments WHERE upload_id=? AND part_number=? ORDER BY ordinal")) {
+            query.setObject(1, id);
+            query.setInt(2, number);
+            try (ResultSet result = query.executeQuery()) {
+                long total = 0;
+                while (result.next()) {
+                    if (result.getInt(1) != segments.size()) throw new IOException("Incomplete multipart manifest");
+                    Segment segment = new Segment((UUID) result.getObject(2), result.getInt(3),
+                        result.getBytes(4), replicaIds(result, 5));
+                    total = Math.addExact(total, segment.length());
+                    segments.add(segment);
+                }
+                if (total != length) throw new IOException("Incomplete multipart manifest");
+            }
+        }
+        return new StoredPart(length, etag, segments);
+    }
+
+    private byte[] readableSegment(Segment segment) {
+        for (UUID id : segment.replicas()) {
+            int index = nodes.index(id);
+            if (index < 0) continue;
+            byte[] bytes = readableReplica(index, segment);
+            if (bytes != null) return bytes;
+        }
+        return null;
+    }
+
     private long currentLength(Connection connection, String bucket, String key) throws SQLException {
         try (PreparedStatement query = connection.prepareStatement("SELECT length FROM cluster_objects WHERE bucket=? AND object_key=?")) {
-            query.setString(1, bucket); query.setString(2, key);
+            query.setString(1, bucket);
+            query.setString(2, key);
             try (ResultSet result = query.executeQuery()) { return result.next() ? result.getLong(1) : -1; }
         }
     }
     private static void bindObject(PreparedStatement update, Metadata data, UUID generation) throws SQLException {
-        update.setString(1, data.bucket()); update.setString(2, data.key()); update.setObject(3, generation);
-        update.setLong(4, data.length()); update.setLong(5, data.modified()); update.setString(6, data.etag());
-        update.setBytes(7, data.sha256()); update.setString(8, data.contentType());
+        update.setString(1, data.bucket());
+        update.setString(2, data.key());
+        update.setObject(3, generation);
+        update.setLong(4, data.length());
+        update.setLong(5, data.modified());
+        update.setString(6, data.etag());
+        update.setBytes(7, data.sha256());
+        update.setString(8, data.contentType());
     }
     private static List<UUID> replicaIds(ResultSet result, int column) throws SQLException, IOException {
         java.sql.Array value = result.getArray(column);
@@ -381,21 +801,28 @@ final class ClusterStore implements ObjectStorage {
         } catch (SQLException error) { return false; }
     }
     RepairReport repairOnce() throws IOException {
-        int scanned = 0, restored = 0, underReplicated = 0, unrecoverable = 0;
+        int scanned = 0, restored = 0, rebalanced = 0, underReplicated = 0, unrecoverable = 0;
         try (Connection reader = connect()) {
             reader.setAutoCommit(false);
+            try (var lock = reader.createStatement()) {
+                lock.execute("SELECT pg_advisory_xact_lock(6834071092782)");
+            }
             try (PreparedStatement query = reader.prepareStatement(
-                "SELECT s.generation, s.ordinal, s.segment_id, s.length, s.sha256, s.replica_ids, s.placement_version FROM cluster_segments s JOIN cluster_objects o ON o.generation=s.generation ORDER BY s.generation, s.ordinal")) {
+                "SELECT s.generation, 0, s.ordinal, s.segment_id, s.length, s.sha256, s.replica_ids, s.placement_version " +
+                "FROM cluster_segments s JOIN cluster_objects o ON o.generation=s.generation " +
+                "UNION ALL SELECT s.upload_id, s.part_number, s.ordinal, s.segment_id, s.length, s.sha256, " +
+                "s.replica_ids, s.placement_version FROM cluster_upload_segments s " +
+                "ORDER BY 1, 2, 3")) {
                 query.setFetchSize(128);
                 try (ResultSet result = query.executeQuery()) {
                     while (result.next()) {
                         scanned++;
                         RepairTarget target = new RepairTarget((UUID) result.getObject(1), result.getInt(2),
-                            result.getLong(7), new Segment((UUID) result.getObject(3), result.getInt(4),
-                                result.getBytes(5), replicaIds(result, 6)));
+                            result.getInt(3), result.getLong(8), new Segment((UUID) result.getObject(4),
+                                result.getInt(5), result.getBytes(6), replicaIds(result, 7)));
                         Segment segment = target.segment();
                         byte[] copy = null;
-                        Set<UUID> healthy = new HashSet<>();
+                        Set<UUID> healthy = new LinkedHashSet<>();
                         Set<UUID> healthyHosts = new HashSet<>();
                         for (UUID id : segment.replicas()) {
                             int node = nodes.index(id);
@@ -406,28 +833,63 @@ final class ClusterStore implements ObjectStorage {
                             healthy.add(id);
                             healthyHosts.add(nodes.faultDomain(node, testNodeDomains));
                         }
-                        if (copy == null) { unrecoverable++; continue; }
+                        if (copy == null) {
+                            unrecoverable++;
+                            continue;
+                        }
+                        List<UUID> preferred = new ArrayList<>();
+                        Set<UUID> preferredHosts = new HashSet<>();
                         for (int node : PlacementPolicy.candidates(segment.id(), nodes, testNodeDomains)) {
                             UUID host = nodes.faultDomain(node, testNodeDomains);
-                            if (healthyHosts.contains(host)) continue;
+                            if (!preferredHosts.add(host)) continue;
+                            preferred.add(nodes.node(node).id());
+                            if (preferred.size() == 3) break;
+                        }
+                        for (UUID id : preferred) {
+                            int node = nodes.index(id);
+                            UUID host = nodes.faultDomain(node, testNodeDomains);
+                            if (healthy.contains(id)) continue;
                             if (repairReplica(node, segment, copy)) {
-                                healthy.add(nodes.node(node).id());
+                                healthy.add(id);
                                 healthyHosts.add(host);
                                 restored++;
                             }
-                            if (healthyHosts.size() == 3) break;
                         }
                         if (healthyHosts.size() < 3) underReplicated++;
-                        Set<UUID> listed = new java.util.LinkedHashSet<>(segment.replicas());
-                        listed.addAll(healthy);
-                        if (listed.size() != segment.replicas().size()) {
+                        List<UUID> listed = new ArrayList<>();
+                        Set<UUID> listedHosts = new HashSet<>();
+                        for (UUID id : preferred) {
+                            if (healthy.contains(id)) {
+                                listed.add(id);
+                                listedHosts.add(nodes.faultDomain(nodes.index(id), testNodeDomains));
+                            }
+                        }
+                        for (UUID id : segment.replicas()) {
+                            if (listed.size() == 3) break;
+                            int node = nodes.index(id);
+                            if (healthy.contains(id) && node >= 0 &&
+                                listedHosts.add(nodes.faultDomain(node, testNodeDomains))) listed.add(id);
+                        }
+                        if (listed.size() < 3) {
+                            for (UUID id : segment.replicas()) {
+                                if (!listed.contains(id)) listed.add(id);
+                            }
+                        }
+                        if (!listed.equals(segment.replicas())) {
+                            String table = target.part() == 0 ? "cluster_segments" : "cluster_upload_segments";
+                            String identity = target.part() == 0 ? "generation=? AND ordinal=?" :
+                                "upload_id=? AND part_number=? AND ordinal=?";
                             try (Connection writer = connect(); PreparedStatement update = writer.prepareStatement(
-                                "UPDATE cluster_segments SET replica_ids=?, placement_version=placement_version+1 WHERE generation=? AND ordinal=? AND placement_version=?")) {
+                                "UPDATE " + table + " SET replica_ids=?, placement_version=placement_version+1 WHERE " +
+                                    identity + " AND placement_version=?")) {
                                 update.setArray(1, writer.createArrayOf("uuid", listed.toArray()));
-                                update.setObject(2, target.generation());
-                                update.setInt(3, target.ordinal());
-                                update.setLong(4, target.version());
-                                update.executeUpdate();
+                                update.setObject(2, target.id());
+                                int next = 3;
+                                if (target.part() != 0) update.setInt(next++, target.part());
+                                update.setInt(next++, target.ordinal());
+                                update.setLong(next, target.version());
+                                if (update.executeUpdate() == 1 && preferred.stream().anyMatch(id ->
+                                    !segment.replicas().contains(id) && listed.contains(id))) rebalanced++;
                             }
                         }
                     }
@@ -435,7 +897,7 @@ final class ClusterStore implements ObjectStorage {
             }
             reader.commit();
         } catch (SQLException error) { throw databaseError(error); }
-        return new RepairReport(scanned, restored, underReplicated, unrecoverable);
+        return new RepairReport(scanned, restored, rebalanced, underReplicated, unrecoverable);
     }
 
     private byte[] readableReplica(int node, Segment segment) {
@@ -450,6 +912,114 @@ final class ClusterStore implements ObjectStorage {
         } catch (IOException unavailable) { return false; }
     }
 
+    GcReport collectGarbage(long minimumAgeMillis, boolean apply) throws IOException {
+        if (minimumAgeMillis < 0 || (minimumAgeMillis == 0 && !testNodeDomains))
+            throw new IllegalArgumentException("Invalid garbage collection age");
+        if (nodes.count() < 2 || nodes.repairTokenUnavailable())
+            throw new IllegalStateException("Garbage collection requires repair authority");
+        int scanned = 0, eligible = 0, deleted = 0, unavailable = 0;
+        try (Connection connection = connect()) {
+            try (var lock = connection.createStatement()) {
+                lock.execute("SELECT pg_advisory_lock(6834071092782)");
+            }
+            lockGc(connection, false);
+            try (PreparedStatement referenced = connection.prepareStatement(
+                "SELECT EXISTS (SELECT 1 FROM cluster_segments s JOIN cluster_objects o " +
+                "ON o.generation=s.generation WHERE s.segment_id=? AND ?=ANY(s.replica_ids) " +
+                "UNION ALL SELECT 1 FROM cluster_upload_segments s " +
+                "WHERE s.segment_id=? AND ?=ANY(s.replica_ids))");
+                 PreparedStatement candidate = connection.prepareStatement(
+                     "SELECT observed_mtime, first_seen FROM cluster_gc_candidates WHERE node_id=? AND segment_id=?");
+                 PreparedStatement mark = connection.prepareStatement(
+                     "INSERT INTO cluster_gc_candidates VALUES (?, ?, ?, ?) ON CONFLICT DO NOTHING");
+                 PreparedStatement reset = connection.prepareStatement(
+                     "UPDATE cluster_gc_candidates SET observed_mtime=?, first_seen=? WHERE node_id=? AND segment_id=?");
+                 PreparedStatement clear = connection.prepareStatement(
+                     "DELETE FROM cluster_gc_candidates WHERE node_id=? AND segment_id=?")) {
+                for (int node = 0; node < nodes.count(); node++) {
+                    boolean reachable = true;
+                    for (int shard = 0; shard < 256 && reachable; shard++) {
+                        String prefix = "%02x".formatted(shard);
+                        UUID after = null;
+                        while (true) {
+                            List<NodeClient.StoredSegment> page;
+                            try { page = nodes.inventory(node, prefix, after); }
+                            catch (IOException error) {
+                                System.err.println("Cluster inventory failed for node " + nodes.node(node).id() +
+                                    ": " + error.getMessage());
+                                unavailable++;
+                                reachable = false;
+                                break;
+                            }
+                            for (NodeClient.StoredSegment segment : page) {
+                                scanned++;
+                                UUID nodeId = nodes.node(node).id();
+                                referenced.setObject(1, segment.id());
+                                referenced.setObject(2, nodeId);
+                                referenced.setObject(3, segment.id());
+                                referenced.setObject(4, nodeId);
+                                try (ResultSet result = referenced.executeQuery()) {
+                                    result.next();
+                                    if (result.getBoolean(1)) {
+                                        if (apply) clearCandidate(clear, nodeId, segment.id());
+                                        continue;
+                                    }
+                                }
+                                eligible++;
+                                if (apply) {
+                                    candidate.setObject(1, nodeId);
+                                    candidate.setObject(2, segment.id());
+                                    long now = System.currentTimeMillis();
+                                    boolean firstObservation = false;
+                                    long firstSeen = now;
+                                    try (ResultSet result = candidate.executeQuery()) {
+                                        if (!result.next()) firstObservation = true;
+                                        else if (result.getLong(1) != segment.modified()) firstObservation = true;
+                                        else firstSeen = result.getLong(2);
+                                    }
+                                    if (firstObservation) {
+                                        reset.setLong(1, segment.modified());
+                                        reset.setLong(2, now);
+                                        reset.setObject(3, nodeId);
+                                        reset.setObject(4, segment.id());
+                                        if (reset.executeUpdate() == 0) {
+                                            mark.setObject(1, nodeId);
+                                            mark.setObject(2, segment.id());
+                                            mark.setLong(3, segment.modified());
+                                            mark.setLong(4, now);
+                                            mark.executeUpdate();
+                                        }
+                                        continue;
+                                    }
+                                    if (now - firstSeen < minimumAgeMillis) continue;
+                                    try {
+                                        if (nodes.deleteOrphan(node, segment, minimumAgeMillis)) deleted++;
+                                        clearCandidate(clear, nodeId, segment.id());
+                                    } catch (IOException error) {
+                                        System.err.println("Cluster deletion failed for segment " + segment.id() +
+                                            ": " + error.getMessage());
+                                        unavailable++;
+                                        reachable = false;
+                                        break;
+                                    }
+                                }
+                            }
+                            if (!reachable || page.size() < 1000) break;
+                            after = page.getLast().id();
+                        }
+                    }
+                }
+            }
+        } catch (SQLException error) { throw databaseError(error); }
+        return new GcReport(scanned, eligible, deleted, unavailable);
+    }
+
+    private static void clearCandidate(PreparedStatement clear, UUID node, UUID segment) throws SQLException {
+        clear.setObject(1, node);
+        clear.setObject(2, segment);
+        clear.executeUpdate();
+    }
+
     @Override public void close() {}
 
     private final class SegmentStream extends InputStream {
@@ -484,6 +1054,9 @@ final class ClusterStore implements ObjectStorage {
             }
             return current.read(buffer, offset, length);
         }
-        @Override public void close() { closed = true; current = null; }
+        @Override public void close() {
+            closed = true;
+            current = null;
+        }
     }
 }

src/cloud/lunarsky/store/DiskStore.java

@@ -34,8 +34,10 @@ final class DiskStore implements ObjectStorage {
 
     DiskStore(Path root, long maxObject, long maxTotal) throws IOException {
         this.root = root;
-        objects = root.resolve("objects"); temporary = root.resolve("pending");
-        this.maxObject = maxObject; this.maxTotal = maxTotal;
+        objects = root.resolve("objects");
+        temporary = root.resolve("pending");
+        this.maxObject = maxObject;
+        this.maxTotal = maxTotal;
         Arrays.setAll(locks, i -> new Object());
         Files.createDirectories(root);
         FileChannel channel = FileChannel.open(root.resolve(".process.lock"), StandardOpenOption.CREATE, StandardOpenOption.WRITE);
@@ -45,7 +47,8 @@ final class DiskStore implements ObjectStorage {
             try { acquired = channel.tryLock(); }
             catch (OverlappingFileLockException e) { throw new IOException("Data directory is already in use", e); }
             if (acquired == null) throw new IOException("Data directory is already in use");
-            Files.createDirectories(objects); Files.createDirectories(temporary);
+            Files.createDirectories(objects);
+            Files.createDirectories(temporary);
             syncDirectory(root);
             try (var paths = Files.list(temporary)) {
                 for (Path p : paths.toList()) if (p.getFileName().toString().endsWith(".part")) Files.delete(p);
@@ -140,12 +143,15 @@ final class DiskStore implements ObjectStorage {
         long count = 0;
         try (OutputStream out = Files.newOutputStream(pending)) {
             out.write(new byte[headerLength]);
-            byte[] buffer = new byte[65536]; int n;
+            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);
+                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");
@@ -200,7 +206,10 @@ final class DiskStore implements ObjectStorage {
             try { input = new DataInputStream(Files.newInputStream(destination)); }
             catch (NoSuchFileException e) { throw new StoreException(404, "NoSuchKey", "Object not found"); }
             try { return new OpenObject(readRecord(input).metadata(), input); }
-            catch (IOException e) { input.close(); throw e; }
+            catch (IOException e) {
+                input.close();
+                throw e;
+            }
         }
     }
 
@@ -242,10 +251,21 @@ final class DiskStore implements ObjectStorage {
                 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, meta)); activePrefix = null; }
+            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, meta));
+                activePrefix = null;
+            }
             lastKey = key;
         }
         return new ListPage(entries, prefixes, truncated ? lastKey : null, truncated);
@@ -256,7 +276,8 @@ final class DiskStore implements ObjectStorage {
         if (magic != MAGIC_V1 && magic != MAGIC_V2) throw new IOException("Invalid object record");
         long length = in.readLong(), modified = in.readLong();
         byte[] md5 = new byte[16], sha = new byte[32];
-        in.readFully(md5); in.readFully(sha);
+        in.readFully(md5);
+        in.readFully(sha);
         if (length < 0) throw new IOException("Invalid object record length");
         if (magic == MAGIC_V1)
             return new Record(new Metadata(length, modified, SigV4.hex(md5), sha,

src/cloud/lunarsky/store/Main.java

@@ -34,8 +34,10 @@ public final class Main {
     }
 
     Main(ObjectStorage store, MultipartStorage multipart, SigV4 authentication, String bucket) {
-        this.store = store; this.multipart = multipart;
-        this.authentication = authentication; this.bucket = bucket;
+        this.store = store;
+        this.multipart = multipart;
+        this.authentication = authentication;
+        this.bucket = bucket;
     }
 
     void handle(HttpExchange exchange) throws IOException {
@@ -58,7 +60,10 @@ public final class Main {
         catch (Exception error) {
             System.err.println("ObjectStore request failed: " + requestId + " " + error.getClass().getSimpleName());
             sendError(exchange, 500, "InternalError", "Storage operation failed", requestId);
-        } finally { if (admitted) slots.release(); exchange.close(); }
+        } finally {
+            if (admitted) slots.release();
+            exchange.close();
+        }
     }
 
     private boolean handleStatus(HttpExchange exchange) throws IOException {
@@ -81,6 +86,16 @@ public final class Main {
     }
 
     private void handleBucket(HttpExchange exchange, Map<String, String> query, String hash) throws IOException {
+        if (exchange.getRequestMethod().equals("GET") && query.containsKey("uploads")) {
+            if (!query.get("uploads").isEmpty() ||
+                !query.keySet().stream().allMatch(java.util.Set.of("uploads", "prefix", "key-marker",
+                    "upload-id-marker", "max-uploads", "x-id")::contains) ||
+                (query.containsKey("x-id") && !"ListMultipartUploads".equals(query.get("x-id"))))
+                unsupported("Bucket operation");
+            requireEmptyBody(exchange, hash);
+            listUploads(exchange, query);
+            return;
+        }
         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) ||
@@ -99,7 +114,8 @@ public final class Main {
         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")) ||
+            ("PutObject".equals(query.get("x-id")) || "CopyObject".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());
@@ -107,9 +123,19 @@ public final class Main {
             handleMultipart(exchange, method, query, key, hash);
             return;
         }
-        if (!method.equals("PUT")) requireEmptyBody(exchange, hash);
+        boolean copy = exchange.getRequestHeaders().containsKey("x-amz-copy-source");
+        if (!method.equals("PUT") && (copy || exchange.getRequestHeaders().containsKey("content-md5") ||
+            exchange.getRequestHeaders().containsKey("x-amz-metadata-directive") ||
+            exchange.getRequestHeaders().containsKey("x-amz-sdk-checksum-algorithm") ||
+            exchange.getRequestHeaders().keySet().stream().anyMatch(name ->
+                name.toLowerCase(Locale.ROOT).startsWith("x-amz-checksum-"))))
+            unsupported("Object upload header");
+        if (!method.equals("PUT") || copy) requireEmptyBody(exchange, hash);
         switch (method) {
-            case "PUT" -> putObject(exchange, key, hash);
+            case "PUT" -> {
+                if (copy) copyObject(exchange, key);
+                else putObject(exchange, key, hash);
+            }
             case "GET", "HEAD" -> readObject(exchange, key);
             case "DELETE" -> deleteObject(exchange, key);
             default -> unsupported("HTTP method");
@@ -120,17 +146,14 @@ public final class Main {
         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))
+                "x-amz-sdk-checksum-algorithm", "x-amz-user-agent", "x-amz-copy-source",
+                "x-amz-metadata-directive").contains(lower) && !lower.startsWith("x-amz-checksum-"))
                 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-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");
+                lower.startsWith("x-amz-copy-source-")) unsupported("Object metadata, encryption, ACL or tagging header");
         }
-        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 {
@@ -141,14 +164,51 @@ public final class Main {
         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");
+        if (headers.containsKey("x-amz-metadata-directive")) unsupported("Copy metadata directive");
+        UploadChecksums checksums = UploadChecksums.from(headers);
         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);
+        ObjectStorage.Metadata data = store.put(bucket, key, checksums.verifying(exchange.getRequestBody()),
+            bytes, hash, checksums.sha256(), condition != null, type);
         exchange.getResponseHeaders().set("ETag", "\"" + data.etag() + "\"");
         exchange.getResponseHeaders().set("x-amz-checksum-sha256", Base64.getEncoder().encodeToString(data.sha256()));
+        checksums.response(exchange.getResponseHeaders());
         exchange.sendResponseHeaders(200, -1);
     }
 
+    private void copyObject(HttpExchange exchange, String key) throws IOException {
+        var headers = exchange.getRequestHeaders();
+        if (headers.containsKey("content-encoding") || headers.containsKey("content-md5") ||
+            headers.containsKey("if-none-match") || headers.containsKey("x-amz-sdk-checksum-algorithm") ||
+            headers.keySet().stream().anyMatch(name -> name.toLowerCase(Locale.ROOT).startsWith("x-amz-checksum-")))
+            unsupported("Copy request header");
+        String source = SigV4.single(headers, "x-amz-copy-source");
+        if (source == null) throw new StoreException(400, "InvalidArgument", "Missing copy source");
+        if (source.startsWith("/")) source = source.substring(1);
+        int separator = source.indexOf('/');
+        if (separator <= 0 || separator == source.length() - 1 || source.indexOf('?') >= 0)
+            throw new StoreException(400, "InvalidArgument", "Invalid copy source");
+        String sourceBucket = SigV4.decode(source.substring(0, separator));
+        String sourceKey = SigV4.decode(source.substring(separator + 1));
+        if (!sourceBucket.equals(bucket)) throw new StoreException(404, "NoSuchBucket", "Bucket not found");
+        if (sourceKey.isEmpty() || sourceKey.getBytes(StandardCharsets.UTF_8).length > 1024 ||
+            sourceKey.indexOf('\0') >= 0)
+            throw new StoreException(400, "InvalidArgument", "Invalid copy source key");
+        String directive = SigV4.single(headers, "x-amz-metadata-directive");
+        if (directive != null && !directive.equals("COPY") && !directive.equals("REPLACE"))
+            throw new StoreException(400, "InvalidArgument", "Invalid metadata directive");
+        if (!"REPLACE".equals(directive) && headers.containsKey("content-type"))
+            unsupported("Content-Type requires REPLACE metadata directive");
+        try (var object = store.open(bucket, sourceKey)) {
+            var sourceMetadata = object.metadata();
+            String type = "REPLACE".equals(directive) ? contentType(headers) : sourceMetadata.contentType();
+            var copied = store.put(bucket, key, object.stream(), sourceMetadata.length(),
+                SigV4.hex(sourceMetadata.sha256()), null, false, type);
+            sendXml(exchange, 200, "<CopyObjectResult><LastModified>" +
+                Instant.ofEpochMilli(copied.modified()) + "</LastModified><ETag>&quot;" +
+                copied.etag() + "&quot;</ETag></CopyObjectResult>");
+        }
+    }
+
     private void deleteObject(HttpExchange exchange, String key) throws IOException {
         if (exchange.getRequestHeaders().containsKey("if-none-match")) unsupported("Conditional delete");
         store.delete(bucket, key);
@@ -178,12 +238,16 @@ public final class Main {
                 query.keySet().stream().allMatch(java.util.Set.of("uploads", "x-id")::contains) &&
                 (!query.containsKey("x-id") || query.get("x-id").equals("CreateMultipartUpload"));
         if (!query.containsKey("uploadId") ||
-            !query.keySet().stream().allMatch(java.util.Set.of("uploadId", "partNumber", "x-id")::contains))
+            !query.keySet().stream().allMatch(java.util.Set.of("uploadId", "partNumber",
+                "part-number-marker", "max-parts", "x-id")::contains))
             return false;
         String xId = query.get("x-id");
         if (method.equals("PUT")) return query.containsKey("partNumber") &&
+            !query.containsKey("part-number-marker") && !query.containsKey("max-parts") &&
             (xId == null || xId.equals("UploadPart"));
         if (query.containsKey("partNumber")) return false;
+        if (method.equals("GET")) return xId == null || xId.equals("ListParts");
+        if (query.containsKey("part-number-marker") || query.containsKey("max-parts")) return false;
         return (method.equals("POST") && (xId == null || xId.equals("CompleteMultipartUpload"))) ||
             (method.equals("DELETE") && (xId == null || xId.equals("AbortMultipartUpload")));
     }
@@ -193,6 +257,12 @@ public final class Main {
         var headers = exchange.getRequestHeaders();
         if (headers.containsKey("content-encoding") || headers.containsKey("if-none-match"))
             unsupported("Multipart request header");
+        if (headers.containsKey("x-amz-copy-source") || headers.containsKey("x-amz-metadata-directive"))
+            unsupported("Multipart copy request");
+        if (!method.equals("PUT") && (headers.containsKey("content-md5") ||
+            headers.containsKey("x-amz-sdk-checksum-algorithm") ||
+            headers.keySet().stream().anyMatch(name -> name.toLowerCase(Locale.ROOT).startsWith("x-amz-checksum-"))))
+            unsupported("Multipart checksum header");
         if (query.containsKey("uploads")) {
             requireEmptyBody(exchange, hash);
             String id = multipart.create(bucket, key, contentType(headers));
@@ -208,9 +278,11 @@ public final class Main {
                 try { number = Integer.parseInt(query.get("partNumber")); }
                 catch (NumberFormatException e) { throw new StoreException(400, "InvalidArgument", "Invalid part number"); }
                 long length = contentLength(headers);
-                String etag = multipart.putPart(id, bucket, key, number, exchange.getRequestBody(), length,
-                    hash, SigV4.single(headers, "x-amz-checksum-sha256"));
+                UploadChecksums checksums = UploadChecksums.from(headers);
+                String etag = multipart.putPart(id, bucket, key, number,
+                    checksums.verifying(exchange.getRequestBody()), length, hash, checksums.sha256());
                 exchange.getResponseHeaders().set("ETag", "\"" + etag + "\"");
+                checksums.response(exchange.getResponseHeaders());
                 exchange.sendResponseHeaders(200, -1);
             }
             case "POST" -> {
@@ -226,10 +298,81 @@ public final class Main {
                 multipart.abort(id, bucket, key);
                 exchange.sendResponseHeaders(204, -1);
             }
+            case "GET" -> {
+                requireEmptyBody(exchange, hash);
+                listParts(exchange, id, key, query);
+            }
             default -> unsupported("Multipart operation");
         }
     }
 
+    private void listParts(HttpExchange exchange, String id, String key, Map<String, String> query) throws IOException {
+        int marker = boundedNumber(query.get("part-number-marker"), 0, 10000, 0);
+        int maxParts = boundedNumber(query.get("max-parts"), 1, 1000, 1000);
+        MultipartStorage.PartPage page = multipart.listParts(id, bucket, key, marker, maxParts);
+        StringBuilder body = new StringBuilder("<ListPartsResult><Bucket>").append(xml(bucket))
+            .append("</Bucket><Key>").append(xml(key)).append("</Key><UploadId>").append(xml(id))
+            .append("</UploadId><PartNumberMarker>").append(marker)
+            .append("</PartNumberMarker><NextPartNumberMarker>").append(page.nextMarker())
+            .append("</NextPartNumberMarker><MaxParts>").append(maxParts)
+            .append("</MaxParts><IsTruncated>").append(page.truncated()).append("</IsTruncated>");
+        for (MultipartStorage.PartInfo part : page.parts()) {
+            body.append("<Part><PartNumber>").append(part.number()).append("</PartNumber><LastModified>")
+                .append(Instant.ofEpochMilli(part.modified())).append("</LastModified><ETag>&quot;")
+                .append(part.etag()).append("&quot;</ETag><Size>").append(part.length()).append("</Size></Part>");
+        }
+        sendXml(exchange, 200, body.append("</ListPartsResult>").toString());
+    }
+
+    private void listUploads(HttpExchange exchange, Map<String, String> query) throws IOException {
+        String prefix = query.getOrDefault("prefix", "");
+        String marker = query.getOrDefault("key-marker", "");
+        String uploadMarker = query.getOrDefault("upload-id-marker", "");
+        if (!uploadMarker.isEmpty() && marker.isEmpty())
+            throw new StoreException(400, "InvalidArgument", "Upload ID marker requires a key marker");
+        int maximum = boundedNumber(query.get("max-uploads"), 1, 1000, 1000);
+        List<MultipartStorage.UploadInfo> uploads = multipart.listUploads(bucket, prefix);
+        List<MultipartStorage.UploadInfo> page = new ArrayList<>();
+        boolean truncated = false;
+        for (MultipartStorage.UploadInfo upload : uploads) {
+            if (upload.key().compareTo(marker) < 0 ||
+                (upload.key().equals(marker) && upload.id().compareTo(uploadMarker) <= 0)) continue;
+            if (page.size() == maximum) {
+                truncated = true;
+                break;
+            }
+            page.add(upload);
+        }
+        StringBuilder body = new StringBuilder("<ListMultipartUploadsResult><Bucket>").append(xml(bucket))
+            .append("</Bucket><KeyMarker>").append(xml(marker)).append("</KeyMarker><UploadIdMarker>")
+            .append(xml(uploadMarker)).append("</UploadIdMarker><MaxUploads>").append(maximum)
+            .append("</MaxUploads><IsTruncated>").append(truncated).append("</IsTruncated>");
+        if (truncated) {
+            MultipartStorage.UploadInfo last = page.getLast();
+            body.append("<NextKeyMarker>").append(xml(last.key())).append("</NextKeyMarker><NextUploadIdMarker>")
+                .append(xml(last.id())).append("</NextUploadIdMarker>");
+        }
+        for (MultipartStorage.UploadInfo upload : page) {
+            body.append("<Upload><Key>").append(xml(upload.key())).append("</Key><UploadId>")
+                .append(xml(upload.id())).append("</UploadId><Initiated>")
+                .append(Instant.ofEpochMilli(upload.created())).append("</Initiated></Upload>");
+        }
+        sendXml(exchange, 200, body.append("</ListMultipartUploadsResult>").toString());
+    }
+
+    private static int boundedNumber(String text, int minimum, int maximum, int defaultValue) {
+        if (text == null) return defaultValue;
+        int value;
+        try {
+            value = Integer.parseInt(text);
+        } catch (NumberFormatException error) {
+            throw new StoreException(400, "InvalidArgument", "Invalid listing limit or marker");
+        }
+        if (value < minimum || value > maximum)
+            throw new StoreException(400, "InvalidArgument", "Invalid listing limit or marker");
+        return value;
+    }
+
     private static long contentLength(com.sun.net.httpserver.Headers headers) {
         String text = SigV4.single(headers, "content-length");
         if (text == null) return -1;
@@ -321,7 +464,8 @@ public final class Main {
             else {
                 object.stream().skipNBytes(range.start());
                 exchange.sendResponseHeaders(status, range.length());
-                byte[] buffer = new byte[65536]; long left = range.length();
+                byte[] buffer = new byte[65536];
+                long left = range.length();
                 while (left > 0) {
                     int n = object.stream().read(buffer, 0, (int) Math.min(buffer.length, left));
                     if (n < 0) throw new IOException("Object body ended before its recorded length");
@@ -345,7 +489,8 @@ public final class Main {
             if (parts[0].isEmpty()) {
                 long suffix = Long.parseLong(parts[1]);
                 if (suffix == 0) throw new NumberFormatException();
-                start = Math.max(0, size - suffix); end = size - 1;
+                start = Math.max(0, size - suffix);
+                end = size - 1;
             } else {
                 start = Long.parseLong(parts[0]);
                 end = parts[1].isEmpty() ? size - 1 : Math.min(Long.parseLong(parts[1]), size - 1);
@@ -482,7 +627,7 @@ public final class Main {
                 required(env, "POSTGRES_PASSWORD"), bucket,
                 java.util.Arrays.stream(urls).map(URI::create).toList(), required(env, "CLUSTER_TOKEN"), null,
                 maxObject, maxTotal, "true".equals(env.get("CLUSTER_TEST_NODE_DOMAINS")));
-            multipart = new UnavailableMultipart();
+            multipart = (ClusterStore) store;
         } else if (mode.equals("disk")) {
             DiskStore disk = new DiskStore(Path.of(env.getOrDefault("DATA_DIR", "/data")), maxObject, maxTotal);
             store = disk;
@@ -494,7 +639,8 @@ public final class Main {
             ? new InetSocketAddress(env.getOrDefault("BIND_ADDRESS", "127.0.0.1"), port)
             : new InetSocketAddress(port), 64);
         var executor = Executors.newVirtualThreadPerTaskExecutor();
-        server.setExecutor(executor); server.createContext("/", app::handle);
+        server.setExecutor(executor);
+        server.createContext("/", app::handle);
         Runtime.getRuntime().addShutdownHook(new Thread(() -> {
             server.stop(5);
             executor.close();
@@ -526,7 +672,10 @@ public final class Main {
         String[] units = {"KiB", "MiB", "GiB", "TiB"};
         double value = bytes;
         int unit = -1;
-        do { value /= 1024; unit++; } while (value >= 1024 && unit < units.length - 1);
+        do {
+            value /= 1024;
+            unit++;
+        } while (value >= 1024 && unit < units.length - 1);
         return String.format(Locale.ROOT, "%.1f %s", value, units[unit]);
     }
     private static String required(Map<String, String> env, String key) {

src/cloud/lunarsky/store/MultipartStorage.java

@@ -6,12 +6,17 @@ import java.util.List;
 
 interface MultipartStorage {
     record Part(int number, String etag) {}
+    record PartInfo(int number, long length, String etag, long modified) {}
+    record PartPage(List<PartInfo> parts, int nextMarker, boolean truncated) {}
+    record UploadInfo(String id, String key, long created) {}
 
     String create(String bucket, String key, String contentType) throws IOException;
     String putPart(String id, String bucket, String key, int number, InputStream input,
                    long length, String expectedHash, String checksum) throws IOException;
     ObjectStorage.Metadata complete(String id, String bucket, String key, List<Part> parts) throws IOException;
     void abort(String id, String bucket, String key) throws IOException;
+    PartPage listParts(String id, String bucket, String key, int marker, int maxParts) throws IOException;
+    List<UploadInfo> listUploads(String bucket, String prefix) throws IOException;
     int activeUploads();
     long stagedBytes();
 }

src/cloud/lunarsky/store/MultipartStore.java

@@ -97,11 +97,14 @@ final class MultipartStore implements MultipartStorage {
             MessageDigest sha = digest("SHA-256"), md5 = digest("MD5");
             long count = 0;
             try (var output = Files.newOutputStream(pending)) {
-                byte[] buffer = new byte[65536]; int n;
+                byte[] buffer = new byte[65536];
+                int n;
                 while ((n = input.read(buffer)) != -1) {
                     count += n;
                     if (count > length) throw new StoreException(413, "EntityTooLarge", "Part exceeds declared size");
-                    sha.update(buffer, 0, n); md5.update(buffer, 0, n); output.write(buffer, 0, n);
+                    sha.update(buffer, 0, n);
+                    md5.update(buffer, 0, n);
+                    output.write(buffer, 0, n);
                 }
             }
             if (count != length) throw new StoreException(400, "IncompleteBody", "Part length does not match Content-Length");
@@ -124,7 +127,8 @@ final class MultipartStore implements MultipartStorage {
             throw new StoreException(400, "InvalidPart", "No valid parts supplied");
         MessageDigest sha = digest("SHA-256");
         List<Path> paths = new ArrayList<>();
-        long total = 0; int last = 0;
+        long total = 0;
+        int last = 0;
         for (Part part : parts) {
             if (part.number() <= last || part.number() > 10000)
                 throw new StoreException(400, "InvalidPartOrder", "Parts must be in ascending order");
@@ -136,8 +140,12 @@ final class MultipartStore implements MultipartStorage {
             if (total > store.maxObject()) throw new StoreException(413, "EntityTooLarge", "Object exceeds the configured size limit");
             MessageDigest md5 = digest("MD5");
             try (var input = Files.newInputStream(file)) {
-                byte[] buffer = new byte[65536]; int n;
-                while ((n = input.read(buffer)) != -1) { sha.update(buffer, 0, n); md5.update(buffer, 0, n); }
+                byte[] buffer = new byte[65536];
+                int n;
+                while ((n = input.read(buffer)) != -1) {
+                    sha.update(buffer, 0, n);
+                    md5.update(buffer, 0, n);
+                }
             }
             if (!SigV4.hex(md5.digest()).equals(part.etag().replace("\"", "")))
                 throw new StoreException(400, "InvalidPart", "Part ETag mismatch");
@@ -156,6 +164,48 @@ final class MultipartStore implements MultipartStorage {
         remove(upload(id, bucket, key));
     }
 
+    @Override public synchronized PartPage listParts(String id, String bucket, String key,
+                                                     int marker, int maxParts) throws IOException {
+        Path dir = upload(id, bucket, key);
+        List<PartInfo> parts = new ArrayList<>();
+        boolean truncated = false;
+        try (var files = Files.list(dir)) {
+            for (Path file : files.filter(path -> path.getFileName().toString().matches("part-[0-9]{5}"))
+                    .sorted().toList()) {
+                int number = Integer.parseInt(file.getFileName().toString().substring(5));
+                if (number <= marker) continue;
+                if (parts.size() == maxParts) {
+                    truncated = true;
+                    break;
+                }
+                MessageDigest md5 = digest("MD5");
+                try (InputStream input = Files.newInputStream(file)) {
+                    byte[] buffer = new byte[65536];
+                    int count;
+                    while ((count = input.read(buffer)) != -1) md5.update(buffer, 0, count);
+                }
+                parts.add(new PartInfo(number, Files.size(file), SigV4.hex(md5.digest()),
+                    Files.getLastModifiedTime(file).toMillis()));
+            }
+        }
+        int next = parts.isEmpty() ? marker : parts.getLast().number();
+        return new PartPage(parts, next, truncated);
+    }
+
+    @Override public synchronized List<UploadInfo> listUploads(String bucket, String prefix) throws IOException {
+        List<UploadInfo> uploads = new ArrayList<>();
+        try (var dirs = Files.list(root)) {
+            for (Path dir : dirs.filter(Files::isDirectory).toList()) {
+                Upload upload = readUpload(dir);
+                if (upload.bucket().equals(bucket) && upload.key().startsWith(prefix))
+                    uploads.add(new UploadInfo(dir.getFileName().toString(), upload.key(),
+                        Files.getLastModifiedTime(dir.resolve("manifest")).toMillis()));
+            }
+        }
+        uploads.sort(Comparator.comparing(UploadInfo::key).thenComparing(UploadInfo::id));
+        return uploads;
+    }
+
     private Path upload(String id, String bucket, String key) throws IOException {
         if (!id.matches("[0-9a-f-]{36}")) throw new StoreException(404, "NoSuchUpload", "Upload not found");
         Path dir = root.resolve(id);
@@ -209,7 +259,8 @@ final class MultipartStore implements MultipartStorage {
                 }
                 int n = current.read(buffer, offset, length);
                 if (n >= 0) return n;
-                current.close(); current = null;
+                current.close();
+                current = null;
             }
         }
         @Override public void close() throws IOException { if (current != null) current.close(); }

src/cloud/lunarsky/store/NodeClient.java

@@ -9,6 +9,7 @@ import java.net.http.HttpResponse;
 import java.security.MessageDigest;
 import java.time.Duration;
 import java.util.HashSet;
+import java.util.ArrayList;
 import java.util.HexFormat;
 import java.util.List;
 import java.util.Set;
@@ -16,6 +17,7 @@ import java.util.UUID;
 
 final class NodeClient {
     record Node(UUID id, UUID hostId, URI url) {}
+    record StoredSegment(UUID id, long modified) {}
 
     private static final HttpClient IDENTITY_HTTP = HttpClient.newBuilder()
         .connectTimeout(Duration.ofSeconds(2)).build();
@@ -42,8 +44,11 @@ final class NodeClient {
     int count() { return nodes.size(); }
     List<Node> nodes() { return nodes; }
     Node node(int index) { return nodes.get(index); }
+    boolean repairTokenUnavailable() { return repairToken == null || repairToken.length() < 32; }
     int index(UUID id) {
-        for (int i = 0; i < nodes.size(); i++) if (nodes.get(i).id().equals(id)) return i;
+        for (int i = 0; i < nodes.size(); i++) {
+            if (nodes.get(i).id().equals(id)) return i;
+        }
         return -1;
     }
     UUID faultDomain(int index, boolean testNodeDomains) {
@@ -141,6 +146,59 @@ final class NodeClient {
         }
     }
 
+    List<StoredSegment> inventory(int index, String shard, UUID after) throws IOException {
+        if (repairToken == null || repairToken.length() < 32)
+            throw new IOException("Repair authority is not available to this process");
+        Node node = nodes.get(index);
+        String path = "/segments?shard=" + shard + (after == null ? "" : "&after=" + after);
+        HttpRequest request = HttpRequest.newBuilder(node.url().resolve(path))
+            .timeout(Duration.ofSeconds(30)).header("X-Cluster-Token", token)
+            .header("X-Cluster-Expected-Node", node.id().toString())
+            .header("X-Cluster-Repair-Token", repairToken).GET().build();
+        HttpResponse<InputStream> response = send(request, HttpResponse.BodyHandlers.ofInputStream());
+        try (InputStream body = response.body()) {
+            if (response.statusCode() != 200) throw new IOException("Node inventory failed: " + response.statusCode());
+            byte[] bytes = body.readNBytes(70001);
+            if (bytes.length > 70000) throw new IOException("Node inventory response is too large");
+            List<StoredSegment> result = new ArrayList<>();
+            String last = after == null ? "" : after.toString();
+            for (String line : new String(bytes, java.nio.charset.StandardCharsets.US_ASCII).split("\n")) {
+                if (line.isEmpty()) continue;
+                String[] fields = line.split(" ", -1);
+                if (fields.length != 2) throw new IOException("Invalid node inventory response");
+                try {
+                    UUID id = UUID.fromString(fields[0]);
+                    if (!id.toString().equals(fields[0]) || !fields[0].startsWith(shard) ||
+                        fields[0].compareTo(last) <= 0)
+                        throw new IOException("Invalid node inventory cursor");
+                    result.add(new StoredSegment(id, Long.parseLong(fields[1])));
+                    last = fields[0];
+                } catch (IllegalArgumentException error) {
+                    throw new IOException("Invalid node inventory response", error);
+                }
+            }
+            if (result.size() > 1000) throw new IOException("Node inventory page is too large");
+            return result;
+        }
+    }
+
+    boolean deleteOrphan(int index, StoredSegment segment, long minimumAgeMillis) throws IOException {
+        if (repairToken == null || repairToken.length() < 32)
+            throw new IOException("Repair authority is not available to this process");
+        Node node = nodes.get(index);
+        HttpRequest request = HttpRequest.newBuilder(node.url().resolve("/segments/" + segment.id()))
+            .timeout(Duration.ofSeconds(30)).header("X-Cluster-Token", token)
+            .header("X-Cluster-Expected-Node", node.id().toString())
+            .header("X-Cluster-Repair-Token", repairToken)
+            .header("X-Cluster-Expected-Mtime", Long.toString(segment.modified()))
+            .header("X-Cluster-Gc-Min-Age-Millis", Long.toString(minimumAgeMillis))
+            .DELETE().build();
+        HttpResponse<Void> response = send(request, HttpResponse.BodyHandlers.discarding());
+        if (response.statusCode() == 204) return true;
+        if (response.statusCode() == 404 || response.statusCode() == 409) return false;
+        throw new IOException("Node refused orphan deletion: " + response.statusCode());
+    }
+
     private <T> HttpResponse<T> send(HttpRequest request, HttpResponse.BodyHandler<T> handler) throws IOException {
         try { return http.send(request, handler); }
         catch (InterruptedException error) {

src/cloud/lunarsky/store/NodeRegistry.java

@@ -52,7 +52,8 @@ final class NodeRegistry {
             connection.commit();
             return new NodeClient.Node(identity.nodeId(), identity.hostId(), url);
         } catch (SQLException | IOException error) {
-            try { connection.rollback(); } catch (SQLException rollback) { error.addSuppressed(rollback); }
+            try { connection.rollback(); }
+            catch (SQLException rollback) { error.addSuppressed(rollback); }
             if (error instanceof IOException io) throw io;
             throw new IOException("Node registration failed", error);
         } finally {
@@ -77,7 +78,8 @@ final class NodeRegistry {
             connection.commit();
             return nodes;
         } catch (SQLException | IOException | RuntimeException error) {
-            try { connection.rollback(); } catch (SQLException rollback) { error.addSuppressed(rollback); }
+            try { connection.rollback(); }
+            catch (SQLException rollback) { error.addSuppressed(rollback); }
             if (error instanceof IOException io) throw io;
             if (error instanceof SQLException sql) throw new IOException("Node registry check failed", sql);
             throw (RuntimeException) error;

src/cloud/lunarsky/store/SchemaMigrator.java

@@ -21,7 +21,7 @@ final class SchemaMigrator {
                     result.next();
                     version = result.getInt(1);
                 }
-                if (version > 2) throw new IOException("Metadata schema is newer than this ObjectStore build");
+                if (version > 4) throw new IOException("Metadata schema is newer than this ObjectStore build");
                 if (version < 1) {
                     statement.execute("CREATE TABLE IF NOT EXISTS cluster_usage (bucket text PRIMARY KEY, used_bytes bigint NOT NULL CHECK (used_bytes >= 0))");
                     statement.execute("CREATE TABLE IF NOT EXISTS cluster_objects (bucket text NOT NULL, object_key text COLLATE \"C\" NOT NULL, generation uuid NOT NULL, length bigint NOT NULL, modified bigint NOT NULL, etag text NOT NULL, sha256 bytea NOT NULL, content_type text NOT NULL, PRIMARY KEY (bucket, object_key))");
@@ -37,6 +37,16 @@ final class SchemaMigrator {
                     statement.execute("CREATE TABLE IF NOT EXISTS cluster_format (singleton integer PRIMARY KEY CHECK (singleton=1), version integer NOT NULL)");
                     statement.execute("INSERT INTO cluster_schema_migrations VALUES (2)");
                 }
+                if (version < 3) {
+                    statement.execute("CREATE TABLE cluster_uploads (upload_id uuid PRIMARY KEY, bucket text NOT NULL, object_key text COLLATE \"C\" NOT NULL, content_type text NOT NULL, created_at bigint NOT NULL)");
+                    statement.execute("CREATE TABLE cluster_upload_parts (upload_id uuid NOT NULL REFERENCES cluster_uploads(upload_id) ON DELETE CASCADE, part_number integer NOT NULL CHECK (part_number BETWEEN 1 AND 10000), length bigint NOT NULL CHECK (length >= 0), etag text NOT NULL, modified bigint NOT NULL, PRIMARY KEY (upload_id, part_number))");
+                    statement.execute("CREATE TABLE cluster_upload_segments (upload_id uuid NOT NULL, part_number integer NOT NULL, ordinal integer NOT NULL, segment_id uuid NOT NULL, length integer NOT NULL, sha256 bytea NOT NULL, replica_ids uuid[] NOT NULL, placement_version bigint NOT NULL DEFAULT 0, PRIMARY KEY (upload_id, part_number, ordinal), FOREIGN KEY (upload_id, part_number) REFERENCES cluster_upload_parts(upload_id, part_number) ON DELETE CASCADE)");
+                    statement.execute("INSERT INTO cluster_schema_migrations VALUES (3)");
+                }
+                if (version < 4) {
+                    statement.execute("CREATE TABLE cluster_gc_candidates (node_id uuid NOT NULL, segment_id uuid NOT NULL, observed_mtime bigint NOT NULL, first_seen bigint NOT NULL, PRIMARY KEY (node_id, segment_id))");
+                    statement.execute("INSERT INTO cluster_schema_migrations VALUES (4)");
+                }
                 statement.execute("INSERT INTO cluster_format SELECT 1, CASE WHEN EXISTS (SELECT 1 FROM cluster_segments WHERE replica_ids IS NULL) THEN 1 ELSE 2 END WHERE NOT EXISTS (SELECT 1 FROM cluster_format)");
             }
             try (PreparedStatement insert = connection.prepareStatement("INSERT INTO cluster_usage VALUES (?, 0) ON CONFLICT DO NOTHING")) {
@@ -60,7 +70,8 @@ final class SchemaMigrator {
             connection.commit();
             return format;
         } catch (SQLException | IOException error) {
-            try { connection.rollback(); } catch (SQLException rollback) { error.addSuppressed(rollback); }
+            try { connection.rollback(); }
+            catch (SQLException rollback) { error.addSuppressed(rollback); }
             if (error instanceof IOException io) throw io;
             throw new IOException("Metadata schema migration failed", error);
         } finally {

src/cloud/lunarsky/store/SigV4.java

@@ -24,7 +24,10 @@ final class SigV4 {
     private final Clock clock;
 
     SigV4(String accessKey, String secretKey, String region, Clock clock) {
-        this.accessKey = accessKey; this.secretKey = secretKey; this.region = region; this.clock = clock;
+        this.accessKey = accessKey;
+        this.secretKey = secretKey;
+        this.region = region;
+        this.clock = clock;
     }
 
     String verify(String method, URI uri, Headers headers) {
@@ -115,13 +118,21 @@ final class SigV4 {
             for (int i = 0; i < value.length();) {
                 if (value.charAt(i) == '%') {
                     if (i + 2 >= value.length()) throw new IllegalArgumentException();
-                    int hi=Character.digit(value.charAt(i+1),16),lo=Character.digit(value.charAt(i+2),16);
+                    int hi = Character.digit(value.charAt(i + 1), 16);
+                    int lo = Character.digit(value.charAt(i + 2), 16);
                     if (hi < 0 || lo < 0) throw new IllegalArgumentException();
-                    bytes.write((hi<<4)|lo);i+=3;
-                }else{int point=value.codePointAt(i);bytes.writeBytes(new String(Character.toChars(point)).getBytes(StandardCharsets.UTF_8));i+=Character.charCount(point);}
+                    bytes.write((hi << 4) | lo);
+                    i += 3;
+                } else {
+                    int point = value.codePointAt(i);
+                    bytes.writeBytes(new String(Character.toChars(point)).getBytes(StandardCharsets.UTF_8));
+                    i += Character.charCount(point);
+                }
             }
             return StandardCharsets.UTF_8.newDecoder().onMalformedInput(java.nio.charset.CodingErrorAction.REPORT).decode(java.nio.ByteBuffer.wrap(bytes.toByteArray())).toString();
-        }catch(IllegalArgumentException|java.nio.charset.CharacterCodingException e){throw new StoreException(400,"InvalidURI","Malformed URI encoding");}
+        } catch (IllegalArgumentException | java.nio.charset.CharacterCodingException e) {
+            throw new StoreException(400, "InvalidURI", "Malformed URI encoding");
+        }
     }
 
     static String encode(String value, boolean keepSlash) {
@@ -146,10 +157,17 @@ final class SigV4 {
         return hmac(hmac(hmac(hmac(("AWS4"+secret).getBytes(StandardCharsets.UTF_8),date),region),"s3"),"aws4_request");
     }
     static byte[] hmac(byte[] key, String text) {
-        try { Mac mac=Mac.getInstance("HmacSHA256");mac.init(new SecretKeySpec(key,"HmacSHA256"));return mac.doFinal(text.getBytes(StandardCharsets.UTF_8)); }
+        try {
+            Mac mac = Mac.getInstance("HmacSHA256");
+            mac.init(new SecretKeySpec(key, "HmacSHA256"));
+            return mac.doFinal(text.getBytes(StandardCharsets.UTF_8));
+        }
         catch (java.security.GeneralSecurityException e) { throw new IllegalStateException(e); }
     }
-    static byte[] hash(byte[] data) { try {return MessageDigest.getInstance("SHA-256").digest(data);}catch(java.security.NoSuchAlgorithmException e){throw new IllegalStateException(e);} }
+    static byte[] hash(byte[] data) {
+        try { return MessageDigest.getInstance("SHA-256").digest(data); }
+        catch (java.security.NoSuchAlgorithmException e) { throw new IllegalStateException(e); }
+    }
     static String hex(byte[] data) { return HexFormat.of().formatHex(data); }
     private static void denied(String message) { throw new StoreException(403,"AccessDenied",message); }
 }

src/cloud/lunarsky/store/UnavailableMultipart.java

@@ -1,17 +0,0 @@
-package cloud.lunarsky.store;
-
-import java.io.InputStream;
-import java.util.List;
-
-final class UnavailableMultipart implements MultipartStorage {
-    private StoreException unavailable() {
-        return new StoreException(501, "NotImplemented", "Multipart uploads are unavailable in the local cluster prototype");
-    }
-    @Override public String create(String bucket, String key, String contentType) { throw unavailable(); }
-    @Override public String putPart(String id, String bucket, String key, int number, InputStream input,
-                                    long length, String expectedHash, String checksum) { throw unavailable(); }
-    @Override public ObjectStorage.Metadata complete(String id, String bucket, String key, List<Part> parts) { throw unavailable(); }
-    @Override public void abort(String id, String bucket, String key) { throw unavailable(); }
-    @Override public int activeUploads() { return 0; }
-    @Override public long stagedBytes() { return 0; }
-}

src/cloud/lunarsky/store/UploadChecksums.java

@@ -0,0 +1,143 @@
+package cloud.lunarsky.store;
+
+import com.sun.net.httpserver.Headers;
+import java.io.FilterInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.Base64;
+import java.util.Locale;
+import java.util.zip.CRC32;
+import java.util.zip.CRC32C;
+import java.util.zip.Checksum;
+
+final class UploadChecksums {
+    private record Algorithm(String name, String header, int length) {}
+
+    private final byte[] contentMd5;
+    private final Algorithm algorithm;
+    private final byte[] expected;
+    private final String encoded;
+
+    private UploadChecksums(byte[] contentMd5, Algorithm algorithm, byte[] expected, String encoded) {
+        this.contentMd5 = contentMd5;
+        this.algorithm = algorithm;
+        this.expected = expected;
+        this.encoded = encoded;
+    }
+
+    static UploadChecksums from(Headers headers) {
+        String md5 = SigV4.single(headers, "content-md5");
+        byte[] contentMd5 = md5 == null ? null : decode(md5, 16);
+        Algorithm algorithm = null;
+        String encoded = null;
+        for (String name : headers.keySet()) {
+            String lower = name.toLowerCase(Locale.ROOT);
+            if (!lower.startsWith("x-amz-checksum-")) continue;
+            if (algorithm != null)
+                throw new StoreException(400, "InvalidRequest", "Supply one checksum algorithm");
+            algorithm = algorithm(lower);
+            encoded = SigV4.single(headers, name);
+        }
+        String selected = SigV4.single(headers, "x-amz-sdk-checksum-algorithm");
+        if (selected != null && (algorithm == null || !selected.equals(algorithm.name())))
+            throw new StoreException(400, "InvalidRequest", "Checksum algorithm and value must match");
+        byte[] expected = algorithm == null ? null : decode(encoded, algorithm.length());
+        return new UploadChecksums(contentMd5, algorithm, expected, encoded);
+    }
+
+    private static Algorithm algorithm(String header) {
+        return switch (header) {
+            case "x-amz-checksum-crc32" -> new Algorithm("CRC32", header, 4);
+            case "x-amz-checksum-crc32c" -> new Algorithm("CRC32C", header, 4);
+            case "x-amz-checksum-sha1" -> new Algorithm("SHA1", header, 20);
+            case "x-amz-checksum-sha256" -> new Algorithm("SHA256", header, 32);
+            case "x-amz-checksum-sha512" -> new Algorithm("SHA512", header, 64);
+            case "x-amz-checksum-md5" -> new Algorithm("MD5", header, 16);
+            default -> throw new StoreException(501, "NotImplemented", "Checksum algorithm is unsupported");
+        };
+    }
+
+    private static byte[] decode(String value, int length) {
+        try {
+            byte[] decoded = Base64.getDecoder().decode(value);
+            if (decoded.length == length) return decoded;
+        } catch (IllegalArgumentException ignored) { }
+        throw new StoreException(400, "InvalidDigest", "Invalid checksum encoding or length");
+    }
+
+    String sha256() {
+        return algorithm != null && algorithm.name().equals("SHA256") ? encoded : null;
+    }
+
+    void response(Headers headers) {
+        if (algorithm != null) headers.set(algorithm.header(), encoded);
+    }
+
+    InputStream verifying(InputStream input) {
+        if (contentMd5 == null && (algorithm == null || algorithm.name().equals("SHA256"))) return input;
+        return new VerifiedInput(input);
+    }
+
+    private final class VerifiedInput extends FilterInputStream {
+        private final MessageDigest md5 = contentMd5 != null ||
+            (algorithm != null && algorithm.name().equals("MD5")) ? digest("MD5") : null;
+        private final MessageDigest hash = algorithm == null ? null : switch (algorithm.name()) {
+            case "SHA1" -> digest("SHA-1");
+            case "SHA512" -> digest("SHA-512");
+            default -> null;
+        };
+        private final Checksum crc = algorithm == null ? null : switch (algorithm.name()) {
+            case "CRC32" -> new CRC32();
+            case "CRC32C" -> new CRC32C();
+            default -> null;
+        };
+        private boolean checked;
+
+        private VerifiedInput(InputStream input) { super(input); }
+
+        @Override public int read() throws IOException {
+            int value = in.read();
+            if (value < 0) verify();
+            else update(new byte[]{(byte) value}, 0, 1);
+            return value;
+        }
+
+        @Override public int read(byte[] bytes, int offset, int length) throws IOException {
+            int count = in.read(bytes, offset, length);
+            if (count < 0) verify();
+            else if (count > 0) update(bytes, offset, count);
+            return count;
+        }
+
+        private void update(byte[] bytes, int offset, int length) {
+            if (md5 != null) md5.update(bytes, offset, length);
+            if (hash != null) hash.update(bytes, offset, length);
+            if (crc != null) crc.update(bytes, offset, length);
+        }
+
+        private void verify() {
+            if (checked) return;
+            checked = true;
+            byte[] actualMd5 = md5 == null ? null : md5.digest();
+            if (contentMd5 != null && !MessageDigest.isEqual(contentMd5, actualMd5))
+                throw new StoreException(400, "BadDigest", "Content-MD5 mismatch");
+            if (algorithm == null || algorithm.name().equals("SHA256")) return;
+            byte[] actual;
+            if (crc != null) {
+                long value = crc.getValue();
+                actual = new byte[]{(byte) (value >>> 24), (byte) (value >>> 16),
+                    (byte) (value >>> 8), (byte) value};
+            } else if (algorithm.name().equals("MD5")) actual = actualMd5;
+            else actual = hash.digest();
+            if (!MessageDigest.isEqual(expected, actual))
+                throw new StoreException(400, "BadDigest", algorithm.name() + " checksum mismatch");
+        }
+    }
+
+    private static MessageDigest digest(String algorithm) {
+        try { return MessageDigest.getInstance(algorithm); }
+        catch (NoSuchAlgorithmException error) { throw new IllegalStateException(error); }
+    }
+}

src/cloud/lunarsky/store/Version.java

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

test/cloud/lunarsky/store/ClusterIntegrationTest.java

@@ -3,8 +3,14 @@ package cloud.lunarsky.store;
 import java.io.ByteArrayInputStream;
 import java.net.URI;
 import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
 import java.util.Arrays;
+import java.util.HexFormat;
+import java.util.List;
 import java.util.Map;
+import java.util.Set;
+import java.util.HashSet;
+import java.util.UUID;
 
 public final class ClusterIntegrationTest {
     private static final String KEY = "cluster-test/survivor";
@@ -21,7 +27,9 @@ public final class ClusterIntegrationTest {
             switch (args[0]) {
                 case "basic" -> {
                     byte[] large = new byte[ClusterNode.MAX_SEGMENT + 37];
-                    for (int i = 0; i < large.length; i++) large[i] = (byte) (i * 31);
+                    for (int i = 0; i < large.length; i++) {
+                        large[i] = (byte) (i * 31);
+                    }
                     String largeKey = "cluster-test/large";
                     put(store, bucket, largeKey, large, false);
                     try (var opened = store.open(bucket, largeKey)) {
@@ -39,7 +47,10 @@ public final class ClusterIntegrationTest {
                             "Overwrite was not visible");
                     }
                     store.delete(bucket, largeKey);
-                    try { store.open(bucket, largeKey); throw new AssertionError("Deleted object remained visible"); }
+                    try {
+                        store.open(bucket, largeKey);
+                        throw new AssertionError("Deleted object remained visible");
+                    }
                     catch (StoreException error) { require(error.status == 404, "Wrong missing-object status"); }
                     put(store, bucket, KEY, stable, false);
                     require(store.ready(), "Healthy cluster is not ready");
@@ -57,13 +68,70 @@ public final class ClusterIntegrationTest {
                     }
                     System.out.println("Cluster degraded test passed");
                 }
+                case "multipart-stage" -> {
+                    String key = "cluster-test/multipart";
+                    String id = store.create(bucket, key, "text/plain");
+                    byte[] first = "hello ".getBytes(StandardCharsets.UTF_8);
+                    byte[] second = "world".getBytes(StandardCharsets.UTF_8);
+                    putPart(store, id, bucket, key, 1, "old".getBytes(StandardCharsets.UTF_8));
+                    putPart(store, id, bucket, key, 1, first);
+                    putPart(store, id, bucket, key, 2, second);
+                    require(store.activeUploads() == 1, "Upload was not retained");
+                    require(store.stagedBytes() == first.length + second.length, "Replaced part was counted twice");
+                    require(store.listUploads(bucket, key).size() == 1, "Upload listing missed the staged upload");
+                    var firstPage = store.listParts(id, bucket, key, 0, 1);
+                    require(firstPage.truncated() && firstPage.parts().size() == 1 && firstPage.nextMarker() == 1,
+                        "Part listing did not paginate");
+                    require(store.listParts(id, bucket, key, 1, 1).parts().getFirst().number() == 2,
+                        "Part marker skipped the second part");
+                    try {
+                        store.open(bucket, key);
+                        throw new AssertionError("Incomplete upload became visible");
+                    } catch (StoreException error) { require(error.status == 404, "Wrong incomplete-upload status"); }
+                    System.out.println("Cluster multipart parts staged and listed");
+                }
+                case "multipart-complete" -> {
+                    String key = "cluster-test/multipart";
+                    var uploads = store.listUploads(bucket, key);
+                    require(uploads.size() == 1, "Upload did not survive gateway restart");
+                    String id = uploads.getFirst().id();
+                    var listed = store.listParts(id, bucket, key, 0, 1000).parts();
+                    require(listed.size() == 2, "Staged parts were lost");
+                    try {
+                        store.complete(id, bucket, key, List.of(new MultipartStorage.Part(1, "0".repeat(32))));
+                        throw new AssertionError("Wrong part ETag was accepted");
+                    } catch (StoreException error) { require(error.status == 400, "Wrong ETag rejection status"); }
+                    var completed = store.complete(id, bucket, key, List.of(
+                        new MultipartStorage.Part(1, listed.get(0).etag()),
+                        new MultipartStorage.Part(2, listed.get(1).etag())));
+                    byte[] expected = "hello world".getBytes(StandardCharsets.UTF_8);
+                    require(completed.length() == expected.length && completed.contentType().equals("text/plain"),
+                        "Completed object metadata is wrong");
+                    MessageDigest digest = MessageDigest.getInstance("MD5");
+                    digest.update(HexFormat.of().parseHex(listed.get(0).etag()));
+                    digest.update(HexFormat.of().parseHex(listed.get(1).etag()));
+                    require(completed.etag().equals(HexFormat.of().formatHex(digest.digest()) + "-2"),
+                        "Multipart ETag is wrong");
+                    try (var opened = store.open(bucket, key)) {
+                        require(Arrays.equals(opened.stream().readAllBytes(), expected), "Completed multipart body is wrong");
+                    }
+                    require(store.activeUploads() == 0 && store.stagedBytes() == 0, "Completed parts still count as staged");
+                    String aborted = store.create(bucket, "cluster-test/aborted", "text/plain");
+                    putPart(store, aborted, bucket, "cluster-test/aborted", 1, expected);
+                    store.abort(aborted, bucket, "cluster-test/aborted");
+                    require(store.activeUploads() == 0 && store.stagedBytes() == 0, "Aborted parts still count as staged");
+                    System.out.println("Cluster multipart completion survived restart and node loss");
+                }
                 case "quorum-lost" -> {
                     require(!store.ready(), "One available node must not be ready");
                     try {
                         put(store, bucket, "cluster-test/rejected", new byte[]{1}, false);
                         throw new AssertionError("Write succeeded with only one node");
                     } catch (StoreException error) { require(error.status == 503, "Wrong unavailable status"); }
-                    try { store.open(bucket, "cluster-test/rejected"); throw new AssertionError("Failed write became visible"); }
+                    try {
+                        store.open(bucket, "cluster-test/rejected");
+                        throw new AssertionError("Failed write became visible");
+                    }
                     catch (StoreException error) { require(error.status == 404, "Partial object became visible"); }
                     System.out.println("Cluster quorum-loss test passed");
                 }
@@ -82,13 +150,18 @@ public final class ClusterIntegrationTest {
                     var executor = java.util.concurrent.Executors.newFixedThreadPool(2);
                     try {
                         var a = executor.submit(() -> {
-                            start.await(); put(store, bucket, key, first, false); return null;
+                            start.await();
+                            put(store, bucket, key, first, false);
+                            return null;
                         });
                         var b = executor.submit(() -> {
-                            start.await(); put(store, bucket, key, second, false); return null;
+                            start.await();
+                            put(store, bucket, key, second, false);
+                            return null;
                         });
                         start.countDown();
-                        a.get(); b.get();
+                        a.get();
+                        b.get();
                         try (var opened = store.open(bucket, key)) {
                             byte[] actual = opened.stream().readAllBytes();
                             require(Arrays.equals(actual, first) || Arrays.equals(actual, second),
@@ -130,6 +203,42 @@ public final class ClusterIntegrationTest {
                     }
                     System.out.println("Joined node accepted new placements while previous objects stayed readable");
                 }
+                case "verify-expanded" -> {
+                    for (int i = 0; i < 32; i++) {
+                        try (var opened = store.open(bucket, "cluster-test/expanded-" + i)) {
+                            require(("expanded object " + i).equals(new String(opened.stream().readAllBytes(),
+                                StandardCharsets.UTF_8)), "Expanded object was lost during maintenance");
+                        }
+                    }
+                    System.out.println("Expanded objects survived repair and cleanup");
+                }
+                case "balanced" -> {
+                    try (var connection = java.sql.DriverManager.getConnection(env.get("POSTGRES_JDBC_URL"),
+                            env.get("POSTGRES_USER"), env.get("POSTGRES_PASSWORD"))) {
+                        NodeClient nodes = NodeRegistry.load(connection,
+                            Arrays.stream(urls).map(URI::create).toList(), env.get("CLUSTER_TOKEN"), null);
+                        int checked = 0;
+                        try (var query = connection.createStatement();
+                             var result = query.executeQuery("SELECT s.segment_id, s.replica_ids FROM cluster_segments s " +
+                                 "JOIN cluster_objects o ON o.generation=s.generation")) {
+                            while (result.next()) {
+                                UUID segment = (UUID) result.getObject(1);
+                                Set<UUID> preferred = new HashSet<>();
+                                Set<UUID> hosts = new HashSet<>();
+                                for (int index : PlacementPolicy.candidates(segment, nodes, true)) {
+                                    if (hosts.add(nodes.faultDomain(index, true))) preferred.add(nodes.node(index).id());
+                                    if (preferred.size() == 3) break;
+                                }
+                                Set<UUID> actual = new HashSet<>();
+                                for (Object id : (Object[]) result.getArray(2).getArray()) actual.add((UUID) id);
+                                require(actual.equals(preferred), "Segment did not move to preferred hosts");
+                                checked++;
+                            }
+                        }
+                        require(checked > 0, "No live segments were checked for placement");
+                    }
+                    System.out.println("Existing segments balanced across preferred hosts");
+                }
                 default -> throw new IllegalArgumentException("Unknown test phase");
             }
         }
@@ -138,6 +247,11 @@ public final class ClusterIntegrationTest {
         store.put(bucket, key, new ByteArrayInputStream(data), data.length, SigV4.hex(SigV4.hash(data)),
             null, createOnly, "application/octet-stream");
     }
+    private static void putPart(ClusterStore store, String id, String bucket, String key, int number, byte[] data)
+            throws Exception {
+        store.putPart(id, bucket, key, number, new ByteArrayInputStream(data), data.length,
+            SigV4.hex(SigV4.hash(data)), null);
+    }
     private static void require(boolean condition, String message) {
         if (!condition) throw new AssertionError(message);
     }

test/cloud/lunarsky/store/ClusterMigrationTest.java

@@ -31,7 +31,9 @@ public final class ClusterMigrationTest {
         if (args[0].equals("create")) {
             UUID segment = UUID.randomUUID();
             byte[] hash = SigV4.hash(DATA);
-            for (int i = 0; i < nodes.count(); i++) nodes.put(i, segment, DATA, hash);
+            for (int i = 0; i < nodes.count(); i++) {
+                nodes.put(i, segment, DATA, hash);
+            }
             try (var connection = DriverManager.getConnection(env.get("POSTGRES_JDBC_URL"),
                     env.get("POSTGRES_USER"), env.get("POSTGRES_PASSWORD"));
                  var statement = connection.createStatement()) {
@@ -40,18 +42,28 @@ public final class ClusterMigrationTest {
                 statement.execute("CREATE TABLE cluster_segments (generation uuid NOT NULL, ordinal integer NOT NULL, segment_id uuid NOT NULL, length integer NOT NULL, sha256 bytea NOT NULL, replicas text NOT NULL, PRIMARY KEY (generation, ordinal))");
                 statement.execute("CREATE TABLE cluster_tombstones (bucket text NOT NULL, object_key text COLLATE \"C\" NOT NULL, generation uuid NOT NULL, deleted_at bigint NOT NULL, PRIMARY KEY (bucket, object_key))");
                 try (var insert = connection.prepareStatement("INSERT INTO cluster_usage VALUES (?, ?)")) {
-                    insert.setString(1, bucket); insert.setLong(2, DATA.length); insert.executeUpdate();
+                    insert.setString(1, bucket);
+                    insert.setLong(2, DATA.length);
+                    insert.executeUpdate();
                 }
                 UUID generation = UUID.randomUUID();
                 try (var insert = connection.prepareStatement("INSERT INTO cluster_objects VALUES (?, ?, ?, ?, ?, ?, ?, ?)")) {
-                    insert.setString(1, bucket); insert.setString(2, KEY); insert.setObject(3, generation);
-                    insert.setLong(4, DATA.length); insert.setLong(5, System.currentTimeMillis());
+                    insert.setString(1, bucket);
+                    insert.setString(2, KEY);
+                    insert.setObject(3, generation);
+                    insert.setLong(4, DATA.length);
+                    insert.setLong(5, System.currentTimeMillis());
                     insert.setString(6, HexFormat.of().formatHex(MessageDigest.getInstance("MD5").digest(DATA)));
-                    insert.setBytes(7, hash); insert.setString(8, "text/plain"); insert.executeUpdate();
+                    insert.setBytes(7, hash);
+                    insert.setString(8, "text/plain");
+                    insert.executeUpdate();
                 }
                 try (var insert = connection.prepareStatement("INSERT INTO cluster_segments VALUES (?, 0, ?, ?, ?, '0,1,2')")) {
-                    insert.setObject(1, generation); insert.setObject(2, segment);
-                    insert.setInt(3, DATA.length); insert.setBytes(4, hash); insert.executeUpdate();
+                    insert.setObject(1, generation);
+                    insert.setObject(2, segment);
+                    insert.setInt(3, DATA.length);
+                    insert.setBytes(4, hash);
+                    insert.executeUpdate();
                 }
             }
             System.out.println("Legacy cluster fixture created");

test/cloud/lunarsky/store/ClusterNodeTest.java

@@ -75,6 +75,28 @@ public final class ClusterNodeTest {
                 client.repair(0, id, value, SigV4.hash(value));
                 require(java.util.Arrays.equals(value, client.get(0, id, value.length, SigV4.hash(value))),
                     "Repair did not restore the original bytes");
+                UUID orphan = UUID.randomUUID();
+                client.put(0, orphan, value, SigV4.hash(value));
+                var inventory = client.inventory(0, orphan.toString().substring(0, 2), null);
+                var listed = inventory.stream().filter(entry -> entry.id().equals(orphan)).findFirst().orElseThrow();
+                var forgedInventory = HttpRequest.newBuilder(uri.resolve("/segments?shard=" +
+                    orphan.toString().substring(0, 2))).header("X-Cluster-Token", token)
+                    .header("X-Cluster-Expected-Node", nodeId.toString()).GET().build();
+                require(HttpClient.newHttpClient().send(forgedInventory, HttpResponse.BodyHandlers.discarding())
+                    .statusCode() == 403, "Gateway token was allowed to list node segments");
+                var forgedDelete = HttpRequest.newBuilder(uri.resolve("/segments/" + orphan))
+                    .header("X-Cluster-Token", token).header("X-Cluster-Expected-Node", nodeId.toString())
+                    .header("X-Cluster-Expected-Mtime", Long.toString(listed.modified()))
+                    .header("X-Cluster-Gc-Min-Age-Millis", "0").DELETE().build();
+                require(HttpClient.newHttpClient().send(forgedDelete, HttpResponse.BodyHandlers.discarding())
+                    .statusCode() == 403, "Gateway token was allowed to delete a segment");
+                require(!client.deleteOrphan(0, new NodeClient.StoredSegment(orphan, listed.modified() - 1), 0),
+                    "Stale inventory entry deleted a segment");
+                require(client.deleteOrphan(0, listed, 0), "Confirmed orphan segment was not deleted");
+                try {
+                    client.get(0, orphan, value.length, SigV4.hash(value));
+                    throw new AssertionError("Deleted orphan was still readable");
+                } catch (IOException expected) { }
             } finally { server.stop(0); }
         }
         Path pending = root.resolve("pending").resolve("unfinished.part");

test/cloud/lunarsky/store/HttpTest.java

@@ -13,10 +13,14 @@ import java.time.Clock;
 import java.time.Instant;
 import java.time.ZoneOffset;
 import java.time.format.DateTimeFormatter;
+import java.util.Base64;
 import java.util.Comparator;
 import java.util.Map;
 import java.util.TreeMap;
 import java.util.concurrent.Executors;
+import java.util.zip.CRC32;
+import java.util.zip.CRC32C;
+import java.util.zip.Checksum;
 
 public final class HttpTest {
     private static final String ACCESS = "TESTACCESSKEY123";
@@ -129,6 +133,106 @@ public final class HttpTest {
             throw new AssertionError("Empty list page failed: " + emptyPage.body());
     }
 
+    private static void testCopy(HttpClient client, String base) throws Exception {
+        String source = "folder/copy source.txt";
+        String target = "folder/copied.txt";
+        byte[] body = "copy source".getBytes(StandardCharsets.UTF_8);
+        status(200, client.send(signedUri(URI.create(base + "/objects/" + SigV4.encode(source, true)),
+            "PUT", body, Map.of("content-type", "text/plain")), HttpResponse.BodyHandlers.ofByteArray()));
+        String header = "/objects/" + SigV4.encode(source, true);
+        var result = client.send(signedUri(URI.create(base + "/objects/" + target), "PUT",
+            new byte[0], Map.of("x-amz-copy-source", header)), HttpResponse.BodyHandlers.ofString());
+        if (result.statusCode() != 200 || !result.body().contains("<CopyObjectResult>") ||
+            !result.body().contains("<ETag>&quot;"))
+            throw new AssertionError("CopyObject failed: " + result.body());
+        var copied = client.send(signed(base, "GET", target, new byte[0]), HttpResponse.BodyHandlers.ofByteArray());
+        status(200, copied);
+        if (!java.util.Arrays.equals(body, copied.body()) ||
+            !"text/plain".equals(copied.headers().firstValue("content-type").orElse("")))
+            throw new AssertionError("Copied body or content type mismatch");
+        status(200, client.send(signedUri(URI.create(base + "/objects/" + target), "PUT",
+            new byte[0], Map.of("x-amz-copy-source", header, "x-amz-metadata-directive", "REPLACE",
+                "content-type", "text/markdown")), HttpResponse.BodyHandlers.ofByteArray()));
+        var replaced = client.send(signed(base, "GET", target, new byte[0]), HttpResponse.BodyHandlers.ofByteArray());
+        status(200, replaced);
+        if (!java.util.Arrays.equals(body, replaced.body()) ||
+            !"text/markdown".equals(replaced.headers().firstValue("content-type").orElse("")))
+            throw new AssertionError("REPLACE metadata directive failed");
+        status(200, client.send(signedUri(URI.create(base + "/objects/" + target), "PUT",
+            new byte[0], Map.of("x-amz-copy-source", "/objects/" + target)),
+            HttpResponse.BodyHandlers.ofByteArray()));
+        status(404, client.send(signedUri(URI.create(base + "/objects/" + target), "PUT",
+            new byte[0], Map.of("x-amz-copy-source", "/objects/missing")),
+            HttpResponse.BodyHandlers.ofByteArray()));
+        status(404, client.send(signedUri(URI.create(base + "/objects/" + target), "PUT",
+            new byte[0], Map.of("x-amz-copy-source", "/other/source")),
+            HttpResponse.BodyHandlers.ofByteArray()));
+        status(400, client.send(signedUri(URI.create(base + "/objects/" + target), "PUT",
+            new byte[0], Map.of("x-amz-copy-source", "/objects/source?versionId=1")),
+            HttpResponse.BodyHandlers.ofByteArray()));
+        status(501, client.send(signedUri(URI.create(base + "/objects/" + target), "PUT",
+            new byte[0], Map.of("x-amz-copy-source", header, "x-amz-metadata-directive", "REPLACE",
+                "content-md5", "AAAAAAAAAAAAAAAAAAAAAA==")), HttpResponse.BodyHandlers.ofByteArray()));
+        status(204, client.send(signed(base, "DELETE", source, new byte[0]),
+            HttpResponse.BodyHandlers.ofByteArray()));
+        status(204, client.send(signed(base, "DELETE", target, new byte[0]),
+            HttpResponse.BodyHandlers.ofByteArray()));
+    }
+
+    private static String encodedChecksum(String algorithm, byte[] body) throws Exception {
+        if (algorithm.startsWith("CRC")) {
+            Checksum checksum = algorithm.equals("CRC32") ? new CRC32() : new CRC32C();
+            checksum.update(body, 0, body.length);
+            long value = checksum.getValue();
+            return Base64.getEncoder().encodeToString(new byte[]{(byte) (value >>> 24),
+                (byte) (value >>> 16), (byte) (value >>> 8), (byte) value});
+        }
+        String name = algorithm.equals("SHA1") ? "SHA-1" :
+            algorithm.equals("SHA256") ? "SHA-256" :
+            algorithm.equals("SHA512") ? "SHA-512" : "MD5";
+        return Base64.getEncoder().encodeToString(java.security.MessageDigest.getInstance(name).digest(body));
+    }
+
+    private static void testChecksums(HttpClient client, String base) throws Exception {
+        URI uri = URI.create(base + "/objects/checksum-target");
+        byte[] body = "checksum payload".getBytes(StandardCharsets.UTF_8);
+        String md5 = encodedChecksum("MD5", body);
+        for (String algorithm : new String[]{"CRC32", "CRC32C", "SHA1", "SHA256", "SHA512", "MD5"}) {
+            String header = "x-amz-checksum-" + algorithm.toLowerCase(java.util.Locale.ROOT);
+            String checksum = encodedChecksum(algorithm, body);
+            var stored = client.send(signedUri(uri, "PUT", body,
+                Map.of("content-md5", md5, header, checksum, "x-amz-sdk-checksum-algorithm", algorithm)),
+                HttpResponse.BodyHandlers.ofByteArray());
+            status(200, stored);
+            if (!checksum.equals(stored.headers().firstValue(header).orElse("")))
+                throw new AssertionError("Missing checksum response: " + algorithm);
+            var bad = client.send(signedUri(uri, "PUT", body,
+                Map.of(header, Base64.getEncoder().encodeToString(new byte[algorithm.startsWith("CRC") ? 4 :
+                    algorithm.equals("SHA1") ? 20 : algorithm.equals("SHA256") ? 32 :
+                    algorithm.equals("SHA512") ? 64 : 16]))), HttpResponse.BodyHandlers.ofString());
+            if (bad.statusCode() != 400 || !bad.body().contains("BadDigest"))
+                throw new AssertionError("Mismatched " + algorithm + " accepted: " + bad.body());
+            var unchanged = client.send(signedUri(uri, "GET", new byte[0], Map.of()),
+                HttpResponse.BodyHandlers.ofByteArray());
+            status(200, unchanged);
+            if (!java.util.Arrays.equals(body, unchanged.body()))
+                throw new AssertionError("Bad checksum replaced stored object");
+        }
+        var badMd5 = client.send(signedUri(uri, "PUT", body,
+            Map.of("content-md5", "AAAAAAAAAAAAAAAAAAAAAA==")), HttpResponse.BodyHandlers.ofString());
+        if (badMd5.statusCode() != 400 || !badMd5.body().contains("BadDigest"))
+            throw new AssertionError("Mismatched Content-MD5 accepted");
+        status(400, client.send(signedUri(uri, "PUT", body,
+            Map.of("content-md5", "invalid")), HttpResponse.BodyHandlers.ofByteArray()));
+        status(400, client.send(signedUri(uri, "PUT", body,
+            Map.of("x-amz-checksum-crc32", encodedChecksum("CRC32", body),
+                "x-amz-sdk-checksum-algorithm", "CRC32C")), HttpResponse.BodyHandlers.ofByteArray()));
+        status(501, client.send(signedUri(uri, "PUT", body,
+            Map.of("x-amz-checksum-crc64nvme", "AAAAAAAAAAA=")), HttpResponse.BodyHandlers.ofByteArray()));
+        status(204, client.send(signedUri(uri, "DELETE", new byte[0], Map.of()),
+            HttpResponse.BodyHandlers.ofByteArray()));
+    }
+
     private static void testMultipart(HttpClient client, String base) throws Exception {
         String movie = "folder/video.mp4";
         URI initiate = URI.create(base + "/objects/" + movie + "?uploads=");
@@ -142,7 +246,23 @@ public final class HttpTest {
             "?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);
+        status(200, partOne);
+        status(200, partTwo);
+        var rejectedPart = client.send(signedUri(URI.create(base + "/objects/" + movie +
+            "?partNumber=1&uploadId=" + upload), "PUT", first,
+            Map.of("content-md5", "AAAAAAAAAAAAAAAAAAAAAA==")), HttpResponse.BodyHandlers.ofString());
+        if (rejectedPart.statusCode() != 400 || !rejectedPart.body().contains("BadDigest"))
+            throw new AssertionError("Mismatched part Content-MD5 accepted");
+        var parts = client.send(signedUri(URI.create(base + "/objects/" + movie +
+            "?uploadId=" + upload + "&max-parts=1"), "GET", new byte[0], Map.of()),
+            HttpResponse.BodyHandlers.ofString());
+        if (parts.statusCode() != 200 || !parts.body().contains("<IsTruncated>true</IsTruncated>") ||
+            !parts.body().contains("<PartNumber>1</PartNumber>"))
+            throw new AssertionError("Multipart part listing failed: " + parts.body());
+        var uploads = client.send(signedUri(URI.create(base + "/objects?uploads&prefix=folder%2F"),
+            "GET", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofString());
+        if (uploads.statusCode() != 200 || !uploads.body().contains("<UploadId>" + upload + "</UploadId>"))
+            throw new AssertionError("Multipart upload listing failed: " + uploads.body());
         String completion = "<CompleteMultipartUpload><Part><PartNumber>1</PartNumber><ETag>" +
             partOne.headers().firstValue("etag").orElseThrow() +
             "</ETag></Part><Part><PartNumber>2</PartNumber><ETag>" +
@@ -155,6 +275,8 @@ public final class HttpTest {
         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");
+        status(404, client.send(signedUri(URI.create(base + "/objects/" + movie + "?uploadId=" + upload),
+            "GET", new byte[0], Map.of()), HttpResponse.BodyHandlers.ofByteArray()));
         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];
@@ -189,9 +311,11 @@ public final class HttpTest {
             HttpClient client = HttpClient.newHttpClient();
             testObjects(client, base);
             testListing(client, base);
+            testCopy(client, base);
+            testChecksums(client, base);
             testMultipart(client, base);
             testDelete(client, base);
-            System.out.println("HTTP tests passed: health, authentication, PUT, GET, HEAD, DELETE, MIME, ranges, listing, multipart");
+            System.out.println("HTTP tests passed: objects, copy, checksums, listing, multipart");
         } finally {
             server.stop(0);
             executor.close();

test/cloud/lunarsky/store/StoreTest.java

@@ -9,25 +9,33 @@ import java.util.List;
 public final class StoreTest {
     interface Operation {void run() throws Exception;}
     static void fails(int status,Operation operation)throws Exception{
-        try{operation.run();throw new AssertionError("Expected "+status);}catch(StoreException error){if(error.status!=status)throw error;}
+        try {
+            operation.run();
+            throw new AssertionError("Expected " + status);
+        } catch (StoreException error) {
+            if (error.status != status) throw error;
+        }
     }
     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");
     }
     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("host","examplebucket.s3.amazonaws.com");
+        headers.set("range","bytes=0-9");
         headers.set("x-amz-date","20130524T000000Z");
         headers.set("x-amz-content-sha256","e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855");
         headers.set("authorization","AWS4-HMAC-SHA256 Credential=AKIAIOSFODNN7EXAMPLE/20130524/us-east-1/s3/aws4_request,SignedHeaders=host;range;x-amz-content-sha256;x-amz-date,Signature=f0e8bdb87c964420e857bd35b5d6ed310bd44f0170aba48dd91039c6036bdb41");
         var clock=java.time.Clock.fixed(java.time.Instant.parse("2013-05-24T00:00:00Z"),java.time.ZoneOffset.UTC);
         var auth=new SigV4("AKIAIOSFODNN7EXAMPLE","wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY","us-east-1",clock);
-        var uri=java.net.URI.create("/test.txt");auth.verify("GET",uri,headers);
+        var uri=java.net.URI.create("/test.txt");
+        auth.verify("GET",uri,headers);
         fails(403,()->auth.verify("GET",java.net.URI.create("/other.txt"),headers));
         fails(403,()->auth.verify("DELETE",uri,headers));
         fails(403,()->new SigV4("AKIAIOSFODNN7EXAMPLE","wrong","us-east-1",clock).verify("GET",uri,headers));
         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));
+        headers.add("host","duplicate");
+        fails(403,()->auth.verify("GET",uri,headers));
         System.out.println("SigV4 official vector and tampering tests passed");
     }
 
@@ -59,7 +67,8 @@ public final class StoreTest {
         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/☾");
+            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");