Release ObjectStore 0.0.8

This commit is contained in:
admin committed 2026-10-10 15:22:21 +02:00
1 parent ed7712a8af
commit d6427fbac9
61 files changed
+7082 -267

No files matched your search

+179
View File
@@ -0,0 +1,179 @@
package cloud.lunarsky.store;
import com.sun.net.httpserver.Headers;
import java.io.ByteArrayInputStream;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
final class Acl {
static final String ALL_USERS = "http://acs.amazonaws.com/groups/global/AllUsers";
static final String AUTHENTICATED_USERS =
"http://acs.amazonaws.com/groups/global/AuthenticatedUsers";
static final int READ = 1;
static final int WRITE = 2;
static final int READ_ACP = 4;
static final int WRITE_ACP = 8;
static final int FULL_CONTROL = READ | WRITE | READ_ACP | WRITE_ACP;
private static final Map<String, Integer> PERMISSIONS = Map.of(
"READ", READ, "WRITE", WRITE, "READ_ACP", READ_ACP,
"WRITE_ACP", WRITE_ACP, "FULL_CONTROL", FULL_CONTROL);
private Acl() {}
static boolean allows(Map<String, String> grants, String principal, String owner, int permission) {
if (owner.equals(principal)) return true;
if (principal != null && (bits(grants.get(principal)) & permission) == permission) return true;
if (principal != null && (bits(grants.get(AUTHENTICATED_USERS)) & permission) == permission)
return true;
return (bits(grants.get(ALL_USERS)) & permission) == permission;
}
static void require(Map<String, String> grants, String principal, String owner, int permission) {
if (!allows(grants, principal, owner, permission))
throw new StoreException(403, "AccessDenied", "Access denied");
}
static Map<String, String> fromHeaders(Headers headers, Set<String> identities) {
String canned = SigV4.single(headers, "x-amz-acl");
Set<String> grantNames = Set.of("x-amz-grant-read", "x-amz-grant-write",
"x-amz-grant-read-acp", "x-amz-grant-write-acp", "x-amz-grant-full-control");
for (String name : headers.keySet()) {
String lower = name.toLowerCase(java.util.Locale.ROOT);
if (lower.startsWith("x-amz-grant-") && !grantNames.contains(lower))
throw new StoreException(501, "NotImplemented", "Unsupported ACL grant header");
if (canned != null && grantNames.contains(lower))
throw new StoreException(400, "InvalidRequest", "Use either a canned ACL or explicit grants");
}
TreeMap<String, Integer> grants = new TreeMap<>();
if (canned != null) {
if (!canned.equals("private") && !canned.equals("public-read") &&
!canned.equals("authenticated-read") &&
!canned.equals("bucket-owner-full-control"))
throw new StoreException(400, "InvalidArgument", "Unsupported canned ACL");
if (canned.equals("public-read")) grants.put(ALL_USERS, READ);
if (canned.equals("authenticated-read")) grants.put(AUTHENTICATED_USERS, READ);
}
for (String name : new String[]{"read", "write", "read-acp", "write-acp", "full-control"}) {
String value = SigV4.single(headers, "x-amz-grant-" + name);
if (value == null) continue;
int permission = PERMISSIONS.get(name.replace('-', '_').toUpperCase(java.util.Locale.ROOT));
for (String grant : value.split(",")) {
String entry = grant.trim();
String principal;
if (entry.startsWith("id=\"") && entry.endsWith("\""))
principal = entry.substring(4, entry.length() - 1);
else if (entry.startsWith("uri=\"") && entry.endsWith("\""))
principal = entry.substring(5, entry.length() - 1);
else throw new StoreException(400, "InvalidArgument", "Invalid ACL grant");
if (!identities.contains(principal) && !principal.equals(ALL_USERS) &&
!principal.equals(AUTHENTICATED_USERS))
throw new StoreException(400, "InvalidArgument", "Unknown ACL grantee");
if (principal.equals(ALL_USERS) && permission != READ)
throw new StoreException(400, "InvalidArgument", "Only public read is supported");
grants.merge(principal, permission, (left, right) -> left | right);
}
}
return encoded(grants);
}
static Map<String, String> validate(Map<String, String> values, Set<String> identities) {
if (values.size() > 64) throw new StoreException(400, "InvalidArgument", "Too many ACL grantees");
TreeMap<String, String> valid = new TreeMap<>();
for (var entry : values.entrySet()) {
String principal = entry.getKey();
int bits = bits(entry.getValue());
if ((!identities.contains(principal) && !principal.equals(ALL_USERS) &&
!principal.equals(AUTHENTICATED_USERS)) ||
bits == 0 || (bits & ~FULL_CONTROL) != 0 ||
principal.equals(ALL_USERS) && bits != READ)
throw new StoreException(400, "InvalidArgument", "Invalid ACL grantee or permission");
valid.put(principal, Integer.toString(bits));
}
return Map.copyOf(valid);
}
static Map<String, String> fromXml(byte[] body, String owner, Set<String> identities) {
try {
var factory = javax.xml.parsers.DocumentBuilderFactory.newInstance();
factory.setNamespaceAware(true);
factory.setFeature("http://apache.org/xml/features/disallow-doctype-decl", true);
factory.setFeature("http://xml.org/sax/features/external-general-entities", false);
factory.setFeature("http://xml.org/sax/features/external-parameter-entities", false);
factory.setFeature(javax.xml.XMLConstants.FEATURE_SECURE_PROCESSING, true);
factory.setExpandEntityReferences(false);
var document = factory.newDocumentBuilder().parse(new ByteArrayInputStream(body));
var root = document.getDocumentElement();
if (!root.getLocalName().equals("AccessControlPolicy")) throw new IllegalArgumentException();
var ownerNodes = root.getElementsByTagNameNS("*", "Owner");
var lists = root.getElementsByTagNameNS("*", "AccessControlList");
if (ownerNodes.getLength() != 1 || lists.getLength() != 1 ||
!owner.equals(text((org.w3c.dom.Element) ownerNodes.item(0), "ID")))
throw new IllegalArgumentException();
TreeMap<String, Integer> grants = new TreeMap<>();
var nodes = ((org.w3c.dom.Element) lists.item(0)).getElementsByTagNameNS("*", "Grant");
if (nodes.getLength() > 64) throw new IllegalArgumentException();
for (int i = 0; i < nodes.getLength(); i++) {
var grant = (org.w3c.dom.Element) nodes.item(i);
var grantees = grant.getElementsByTagNameNS("*", "Grantee");
if (grantees.getLength() != 1) throw new IllegalArgumentException();
var grantee = (org.w3c.dom.Element) grantees.item(0);
String type = grantee.getAttributeNS("http://www.w3.org/2001/XMLSchema-instance", "type");
String principal = switch (type) {
case "CanonicalUser" -> text(grantee, "ID");
case "Group" -> text(grantee, "URI");
default -> throw new IllegalArgumentException();
};
Integer permission = PERMISSIONS.get(text(grant, "Permission"));
if (permission == null) throw new IllegalArgumentException();
grants.merge(principal, permission, (left, right) -> left | right);
}
grants.remove(owner);
return validate(encoded(grants), identities);
} catch (Exception error) {
throw new StoreException(400, "MalformedACLError", "Invalid access control policy");
}
}
private static String text(org.w3c.dom.Element element, String name) {
var nodes = element.getElementsByTagNameNS("*", name);
if (nodes.getLength() != 1) throw new IllegalArgumentException();
return nodes.item(0).getTextContent().trim();
}
static String xml(Map<String, String> grants, String owner) {
StringBuilder xml = new StringBuilder("<AccessControlPolicy xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\"><Owner><ID>")
.append(owner).append("</ID></Owner><AccessControlList>");
grant(xml, owner, "FULL_CONTROL", false);
for (var entry : new TreeMap<>(grants).entrySet()) {
if (entry.getKey().equals(owner)) continue;
int bits = bits(entry.getValue());
if (bits == FULL_CONTROL) grant(xml, entry.getKey(), "FULL_CONTROL",
entry.getKey().equals(ALL_USERS) || entry.getKey().equals(AUTHENTICATED_USERS));
else for (var permission : PERMISSIONS.entrySet())
if (!permission.getKey().equals("FULL_CONTROL") && (bits & permission.getValue()) != 0)
grant(xml, entry.getKey(), permission.getKey(),
entry.getKey().equals(ALL_USERS) || entry.getKey().equals(AUTHENTICATED_USERS));
}
return xml.append("</AccessControlList></AccessControlPolicy>").toString();
}
private static void grant(StringBuilder xml, String principal, String permission, boolean group) {
xml.append("<Grant><Grantee xmlns:xsi=\"http://www.w3.org/2001/XMLSchema-instance\" xsi:type=\"")
.append(group ? "Group\"><URI>" : "CanonicalUser\"><ID>")
.append(principal).append(group ? "</URI>" : "</ID>")
.append("</Grantee><Permission>").append(permission).append("</Permission></Grant>");
}
private static Map<String, String> encoded(Map<String, Integer> grants) {
TreeMap<String, String> values = new TreeMap<>();
grants.forEach((key, value) -> values.put(key, Integer.toString(value)));
return Map.copyOf(values);
}
private static int bits(String value) {
if (value == null) return 0;
try { return Integer.parseInt(value); }
catch (NumberFormatException error) { return 0; }
}
}
@@ -0,0 +1,175 @@
package cloud.lunarsky.store;
import java.io.FilterInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.util.Base64;
import java.util.HexFormat;
import java.util.zip.CRC32;
import java.util.zip.CRC32C;
import java.util.zip.Checksum;
final class AwsChunkedInputStream extends FilterInputStream {
private static final String EMPTY_HASH = SigV4.hex(SigV4.hash(new byte[0]));
private final SigV4.Verified authorization;
private final long decodedLength;
private final String trailerName;
private final MessageDigest chunkHash;
private final MessageDigest trailerHash;
private final Checksum trailerCrc;
private final XxHashes trailerXxhash;
private long decoded;
private long chunkLeft;
private String suppliedSignature;
private String previousSignature;
private boolean finished;
private String trailerValue;
AwsChunkedInputStream(InputStream input, SigV4.Verified authorization,
long decodedLength, String trailerName) {
super(input);
this.authorization = authorization;
this.decodedLength = decodedLength;
this.trailerName = trailerName;
this.previousSignature = authorization.signature();
try { this.chunkHash = MessageDigest.getInstance("SHA-256"); }
catch (java.security.NoSuchAlgorithmException error) { throw new IllegalStateException(error); }
this.trailerCrc = trailerName == null ? null : switch (trailerName) {
case "x-amz-checksum-crc32" -> new CRC32();
case "x-amz-checksum-crc32c" -> new CRC32C();
case "x-amz-checksum-crc64nvme" -> new Crc64Nvme();
default -> null;
};
this.trailerXxhash = trailerName != null && java.util.Set.of("x-amz-checksum-xxhash64",
"x-amz-checksum-xxhash3", "x-amz-checksum-xxhash128").contains(trailerName)
? new XxHashes(trailerName.substring("x-amz-checksum-".length()).toUpperCase(java.util.Locale.ROOT))
: null;
String algorithm = trailerName == null ? null : switch (trailerName) {
case "x-amz-checksum-sha1" -> "SHA-1";
case "x-amz-checksum-sha256" -> "SHA-256";
case "x-amz-checksum-sha512" -> "SHA-512";
case "x-amz-checksum-md5" -> "MD5";
default -> null;
};
if (trailerName != null && trailerCrc == null && trailerXxhash == null && algorithm == null)
throw new StoreException(501, "NotImplemented", "Checksum trailer is unsupported");
try { this.trailerHash = algorithm == null ? null : MessageDigest.getInstance(algorithm); }
catch (java.security.NoSuchAlgorithmException error) { throw new IllegalStateException(error); }
}
String trailerValue() { return trailerValue; }
@Override public int read() throws IOException {
byte[] one = new byte[1];
int count;
do {
count = read(one, 0, 1);
} while (count == 0);
return count < 0 ? -1 : one[0] & 255;
}
@Override public int read(byte[] bytes, int offset, int length) throws IOException {
java.util.Objects.checkFromIndexSize(offset, length, bytes.length);
if (length == 0) return 0;
if (finished) return -1;
if (chunkLeft == 0) nextChunk();
if (finished) return -1;
int count = in.read(bytes, offset, (int) Math.min(length, chunkLeft));
if (count < 0) throw invalid("Incomplete signed chunk");
if (count == 0) return 0;
chunkHash.update(bytes, offset, count);
if (trailerCrc != null) trailerCrc.update(bytes, offset, count);
if (trailerHash != null) trailerHash.update(bytes, offset, count);
if (trailerXxhash != null) trailerXxhash.update(bytes, offset, count);
chunkLeft -= count;
decoded += count;
if (chunkLeft == 0) {
if (!line().isEmpty()) throw invalid("Missing signed chunk separator");
finishChunk();
}
return count;
}
private void nextChunk() throws IOException {
String line = line();
int separator = line.indexOf(";chunk-signature=");
if (separator < 1 || separator != line.lastIndexOf(";chunk-signature="))
throw invalid("Invalid signed chunk header");
String size = line.substring(0, separator);
suppliedSignature = line.substring(separator + 17);
if (!size.matches("[0-9a-fA-F]{1,16}") || !suppliedSignature.matches("[0-9a-f]{64}"))
throw invalid("Invalid signed chunk header");
try { chunkLeft = Long.parseUnsignedLong(size, 16); }
catch (NumberFormatException error) { throw invalid("Invalid signed chunk size"); }
if (chunkLeft > decodedLength - decoded) throw invalid("Signed chunks exceed decoded length");
chunkHash.reset();
if (chunkLeft == 0) {
finishChunk();
if (decoded != decodedLength) throw invalid("Decoded length mismatch");
if (trailerName == null) {
if (!line().isEmpty()) throw invalid("Invalid signed chunk ending");
} else {
String trailer = line();
if (!trailer.startsWith(trailerName + ":")) throw invalid("Missing signed checksum trailer");
trailerValue = trailer.substring(trailerName.length() + 1);
byte[] actual;
if (trailerCrc != null) {
long value = trailerCrc.getValue();
actual = new byte[trailerName.equals("x-amz-checksum-crc64nvme") ? 8 : 4];
for (int i = actual.length - 1; i >= 0; i--) {
actual[i] = (byte) value;
value >>>= 8;
}
} else actual = trailerXxhash != null ? trailerXxhash.digest() : trailerHash.digest();
if (!Base64.getEncoder().encodeToString(actual).equals(trailerValue))
throw new StoreException(400, "BadDigest", "Checksum trailer mismatch");
String signature = line();
if (!signature.matches("x-amz-trailer-signature=[0-9a-f]{64}"))
throw invalid("Missing trailer signature");
String toSign = "AWS4-HMAC-SHA256-TRAILER\n" + authorization.date() + "\n" +
authorization.scope() + "\n" + previousSignature + "\n" +
SigV4.hex(SigV4.hash((trailerName + ":" + trailerValue + "\n")
.getBytes(StandardCharsets.UTF_8)));
String expected = SigV4.hex(SigV4.hmac(authorization.signingKey(), toSign));
if (!MessageDigest.isEqual(expected.getBytes(StandardCharsets.US_ASCII),
signature.substring(24).getBytes(StandardCharsets.US_ASCII)))
throw invalid("Trailer signature mismatch");
if (!line().isEmpty()) throw invalid("Invalid trailer ending");
}
if (in.read() != -1) throw invalid("Extra bytes after signed payload");
finished = true;
}
}
private void finishChunk() throws IOException {
String toSign = "AWS4-HMAC-SHA256-PAYLOAD\n" + authorization.date() + "\n" +
authorization.scope() + "\n" + previousSignature + "\n" + EMPTY_HASH + "\n" +
SigV4.hex(chunkHash.digest());
byte[] expected = SigV4.hmac(authorization.signingKey(), toSign);
if (!MessageDigest.isEqual(expected, HexFormat.of().parseHex(suppliedSignature)))
throw invalid("Signed chunk signature mismatch");
previousSignature = suppliedSignature;
}
private String line() throws IOException {
byte[] bytes = new byte[512];
int count = 0;
while (count < bytes.length) {
int value = in.read();
if (value < 0) throw invalid("Incomplete signed chunk framing");
if (value == '\r') {
if (in.read() != '\n') throw invalid("Invalid signed chunk line ending");
return new String(bytes, 0, count, StandardCharsets.US_ASCII);
}
if (value < 32 || value > 126) throw invalid("Invalid signed chunk line");
bytes[count++] = (byte) value;
}
throw invalid("Signed chunk header is too long");
}
private static StoreException invalid(String message) {
return new StoreException(400, "InvalidRequest", message);
}
}
+12 -1
View File
@@ -79,6 +79,15 @@ public final class Cli {
try (var paths = Files.walk(objects)) {
for (Path path : paths.filter(Files::isRegularFile).toList()) inspectObject(objects, path, verify, report);
}
Path versions = root.resolve("versions");
if (Files.isDirectory(versions)) {
try (var paths = Files.walk(versions)) {
for (Path path : paths.filter(Files::isRegularFile).toList()) {
if (!path.getFileName().toString().equals("manifest"))
inspectObject(versions, path, verify, report);
}
}
}
Path multipart = root.resolve("multipart");
if (Files.isDirectory(multipart)) {
try (var uploads = Files.list(multipart)) {
@@ -100,7 +109,9 @@ public final class Cli {
if (meta.key() == null) report.legacyObjects++;
else {
String id = SigV4.hex(SigV4.hash((meta.bucket() + "/" + meta.key()).getBytes(StandardCharsets.UTF_8)));
Path expected = objects.resolve(id.substring(0, 2)).resolve(id);
Path expected = objects.getFileName().toString().equals("versions")
? objects.resolve(id.substring(0, 2)).resolve(id).resolve(path.getFileName())
: objects.resolve(id.substring(0, 2)).resolve(id);
if (!path.equals(expected)) report.problem("Mismatched object path: " + path);
}
if (size - record.headerLength() != meta.length()) {
+251
View File
@@ -0,0 +1,251 @@
package cloud.lunarsky.store;
import com.sun.net.httpserver.HttpExchange;
import java.io.FilterInputStream;
import java.io.FilterOutputStream;
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
final class ClientLimits {
private static final int MAX_CLIENTS = 10_000;
private static final long IDLE_NANOS = 300_000_000_000L;
private final int requestsPerSecond;
private final int requestBurst;
private final long bytesPerSecond;
private final long byteBurst;
private final int maxInFlight;
private final Set<String> trustedProxies;
private final Map<String, Client> clients = new HashMap<>();
private long admissions;
private ClientLimits(int requestsPerSecond, int requestBurst, long bytesPerSecond,
long byteBurst, int maxInFlight, Set<String> trustedProxies) {
this.requestsPerSecond = requestsPerSecond;
this.requestBurst = requestBurst;
this.bytesPerSecond = bytesPerSecond;
this.byteBurst = byteBurst;
this.maxInFlight = maxInFlight;
this.trustedProxies = trustedProxies;
}
static ClientLimits disabled() {
return new ClientLimits(0, 0, 0, 0, 0, Set.of());
}
static ClientLimits fromEnvironment(Map<String, String> environment) {
int requests = number(environment, "PUBLIC_REQUESTS_PER_SECOND", 0);
int requestBurst = number(environment, "PUBLIC_REQUEST_BURST", requests);
long bytes = longNumber(environment, "PUBLIC_BYTES_PER_SECOND", 0);
long byteBurst = longNumber(environment, "PUBLIC_BYTE_BURST", bytes);
int inFlight = number(environment, "PUBLIC_MAX_IN_FLIGHT_PER_IP",
requests > 0 || bytes > 0 ? 8 : 0);
if (requests < 0 || requestBurst < 0 || bytes < 0 || byteBurst < 0 || inFlight < 0 ||
requests > 0 && requestBurst < 1 || requests == 0 && requestBurst != 0 ||
bytes > 0 && (byteBurst < 1 || byteBurst > 1_073_741_824L) ||
bytes == 0 && byteBurst != 0 ||
(requests > 0 || bytes > 0) && inFlight < 1)
throw new IllegalArgumentException("Invalid public client limits");
Set<String> proxies = new HashSet<>();
String configured = environment.getOrDefault("PUBLIC_TRUSTED_PROXY_IPS", "").trim();
if (!configured.isEmpty()) {
for (String item : configured.split(",", -1))
proxies.add(numericAddress(item.trim()).getHostAddress());
}
if (requests == 0 && bytes == 0 && inFlight == 0 && !proxies.isEmpty())
throw new IllegalArgumentException("Trusted proxy IPs require public client limits");
return new ClientLimits(requests, requestBurst, bytes, byteBurst, inFlight, Set.copyOf(proxies));
}
private static int number(Map<String, String> environment, String name, int fallback) {
String value = environment.get(name);
if (value == null || value.isBlank()) return fallback;
try { return Integer.parseInt(value); }
catch (NumberFormatException error) { throw new IllegalArgumentException("Invalid " + name, error); }
}
private static long longNumber(Map<String, String> environment, String name, long fallback) {
String value = environment.get(name);
if (value == null || value.isBlank()) return fallback;
try { return Long.parseLong(value); }
catch (NumberFormatException error) { throw new IllegalArgumentException("Invalid " + name, error); }
}
private static InetAddress numericAddress(String value) {
try {
if (value.matches("[0-9]{1,3}(\\.[0-9]{1,3}){3}")) {
String[] parts = value.split("\\.");
byte[] octets = new byte[4];
for (int i = 0; i < 4; i++) {
int octet = Integer.parseInt(parts[i]);
if (octet > 255) throw new IllegalArgumentException("Invalid IP address");
octets[i] = (byte) octet;
}
return InetAddress.getByAddress(octets);
}
if (value.contains(":") && value.matches("[0-9A-Fa-f:.]+"))
return InetAddress.getByName(value);
} catch (UnknownHostException error) {
throw new IllegalArgumentException("Invalid IP address", error);
}
throw new IllegalArgumentException("Expected numeric IP address");
}
private String address(HttpExchange exchange) {
InetAddress peer = exchange.getRemoteAddress().getAddress();
String peerAddress = peer.getHostAddress();
if (!trustedProxies.contains(peerAddress)) return peerAddress;
var values = exchange.getRequestHeaders().get("X-Real-IP");
if (values == null || values.size() != 1)
throw new StoreException(400, "InvalidRequest", "Trusted proxy must supply one X-Real-IP address");
try { return numericAddress(values.getFirst()).getHostAddress(); }
catch (IllegalArgumentException error) {
throw new StoreException(400, "InvalidRequest", "Trusted proxy supplied an invalid client address");
}
}
Client enter(HttpExchange exchange) {
if (requestsPerSecond == 0 && bytesPerSecond == 0 && maxInFlight == 0) return null;
String path = exchange.getRequestURI().getRawPath();
if (exchange.getRemoteAddress().getAddress().isLoopbackAddress() &&
exchange.getRequestHeaders().get("X-Real-IP") == null &&
path.equals("/health")) return null;
String address = address(exchange);
Client client;
synchronized (this) {
long now = System.nanoTime();
if (++admissions % 1024 == 0 || clients.size() >= MAX_CLIENTS)
clients.entrySet().removeIf(entry -> entry.getValue().inFlight == 0 &&
now - entry.getValue().lastSeen > IDLE_NANOS);
client = clients.get(address);
if (client == null) {
if (clients.size() >= MAX_CLIENTS)
throw new StoreException(503, "SlowDown", "Client limit table is full");
client = new Client(now, requestBurst, byteBurst);
clients.put(address, client);
}
refill(client, now);
client.lastSeen = now;
if (maxInFlight > 0 && client.inFlight >= maxInFlight)
throw new StoreException(503, "SlowDown", "Too many concurrent requests from this client");
if (requestsPerSecond > 0 && client.requestTokens < 1)
throw new StoreException(503, "SlowDown", "Client request rate exceeded");
if (requestsPerSecond > 0) client.requestTokens--;
client.inFlight++;
}
if (bytesPerSecond > 0) {
try {
exchange.setStreams(new LimitedInput(exchange.getRequestBody(), client),
new LimitedOutput(exchange.getResponseBody(), client));
} catch (RuntimeException error) {
leave(client);
throw error;
}
}
return client;
}
synchronized void leave(Client client) {
if (client != null) {
client.inFlight--;
client.lastSeen = System.nanoTime();
}
}
private void refill(Client client, long now) {
double seconds = Math.max(0, now - client.lastRefill) / 1_000_000_000.0;
if (requestsPerSecond > 0)
client.requestTokens = Math.min(requestBurst, client.requestTokens + seconds * requestsPerSecond);
if (bytesPerSecond > 0)
client.byteTokens = Math.min(byteBurst, client.byteTokens + seconds * bytesPerSecond);
client.lastRefill = now;
}
private void pace(Client client, int count) throws IOException {
while (true) {
long wait;
synchronized (this) {
refill(client, System.nanoTime());
if (client.byteTokens >= count) {
client.byteTokens -= count;
return;
}
wait = Math.max(1_000_000L,
(long) Math.ceil((count - client.byteTokens) * 1_000_000_000.0 / bytesPerSecond));
}
try { Thread.sleep(Math.min(wait / 1_000_000L + 1, 1000)); }
catch (InterruptedException error) {
Thread.currentThread().interrupt();
throw new IOException("Transfer interrupted while waiting for client bandwidth", error);
}
}
}
private int chunk() { return (int) Math.min(16_384, byteBurst); }
static final class Client {
private long lastSeen;
private long lastRefill;
private double requestTokens;
private double byteTokens;
private int inFlight;
private Client(long now, int requestBurst, long byteBurst) {
lastSeen = now;
lastRefill = now;
requestTokens = requestBurst;
byteTokens = byteBurst;
}
}
private final class LimitedInput extends FilterInputStream {
private final Client client;
private LimitedInput(java.io.InputStream input, Client client) {
super(input);
this.client = client;
}
@Override public int read() throws IOException {
int value = in.read();
if (value >= 0) pace(client, 1);
return value;
}
@Override public int read(byte[] bytes, int offset, int length) throws IOException {
int count = in.read(bytes, offset, Math.min(length, chunk()));
if (count > 0) pace(client, count);
return count;
}
}
private final class LimitedOutput extends FilterOutputStream {
private final Client client;
private LimitedOutput(java.io.OutputStream output, Client client) {
super(output);
this.client = client;
}
@Override public void write(int value) throws IOException {
pace(client, 1);
out.write(value);
}
@Override public void write(byte[] bytes, int offset, int length) throws IOException {
java.util.Objects.checkFromIndexSize(offset, length, bytes.length);
int left = length;
while (left > 0) {
int count = Math.min(left, chunk());
pace(client, count);
out.write(bytes, offset, count);
offset += count;
left -= count;
}
}
}
}
+641 -90
View File
@@ -23,13 +23,15 @@ import java.util.HexFormat;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
final class ClusterStore implements ObjectStorage, MultipartStorage {
private record Segment(UUID id, int length, byte[] hash, List<UUID> replicas) {}
private record RepairTarget(UUID id, int part, int ordinal, long version, Segment segment) {}
private record Upload(String contentType) {}
private record Upload(String contentType, Map<String, String> userMetadata,
Map<String, String> tags, Map<String, String> acl) {}
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) {}
@@ -59,6 +61,168 @@ final class ClusterStore implements ObjectStorage, MultipartStorage {
private Connection connect() throws SQLException { return DriverManager.getConnection(jdbcUrl, user, password); }
@Override public Limits limits() { return new Limits(maxObject, maxTotal); }
@Override public void ensureBucket(String bucket) throws IOException {
try { bucket(bucket); }
catch (StoreException error) {
if (error.status != 404) throw error;
try { createBucket(bucket); }
catch (StoreException created) {
if (created.status != 409) throw created;
}
}
}
@Override public Bucket bucket(String name) throws IOException {
try (Connection connection = connect(); PreparedStatement query = connection.prepareStatement(
"SELECT created_at, versioning_state, acl FROM cluster_buckets WHERE name=?")) {
query.setString(1, name);
try (ResultSet result = query.executeQuery()) {
if (!result.next()) throw new StoreException(404, "NoSuchBucket", "Bucket not found");
return new Bucket(name, result.getLong(1), VersioningState.valueOf(result.getString(2)),
ObjectAttributes.decode(result.getBytes(3)));
}
} catch (SQLException error) { throw databaseError(error); }
}
@Override public List<Bucket> buckets() throws IOException {
List<Bucket> result = new ArrayList<>();
try (Connection connection = connect(); var query = connection.createStatement();
ResultSet rows = query.executeQuery("SELECT name, created_at, versioning_state, acl FROM cluster_buckets ORDER BY name")) {
while (rows.next()) result.add(new Bucket(rows.getString(1), rows.getLong(2),
VersioningState.valueOf(rows.getString(3)), ObjectAttributes.decode(rows.getBytes(4))));
return result;
} catch (SQLException error) { throw databaseError(error); }
}
@Override public void setVersioning(String bucket, VersioningState state) throws IOException {
if (state == VersioningState.NEVER)
throw new StoreException(400, "InvalidArgument", "Versioning cannot be disabled after it is enabled");
try (Connection connection = connect()) {
connection.setAutoCommit(false);
try {
lockUsage(connection, bucket);
VersioningState old = versioningState(connection, bucket);
if (old == VersioningState.NEVER && state == VersioningState.SUSPENDED)
throw new StoreException(400, "InvalidArgument", "Enable versioning before suspending it");
try (PreparedStatement update = connection.prepareStatement(
"UPDATE cluster_buckets SET versioning_state=? WHERE name=?")) {
update.setString(1, state.name());
update.setString(2, bucket);
update.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 void setBucketAcl(String bucket, Map<String, String> acl) throws IOException {
try (Connection connection = connect(); PreparedStatement update = connection.prepareStatement(
"UPDATE cluster_buckets SET acl=? WHERE name=?")) {
update.setBytes(1, ObjectAttributes.encode(acl, 2048));
update.setString(2, bucket);
if (update.executeUpdate() == 0)
throw new StoreException(404, "NoSuchBucket", "Bucket not found");
} catch (SQLException error) { throw databaseError(error); }
}
private static VersioningState versioningState(Connection connection, String bucket) throws SQLException {
try (PreparedStatement query = connection.prepareStatement(
"SELECT versioning_state FROM cluster_buckets WHERE name=? FOR UPDATE")) {
query.setString(1, bucket);
try (ResultSet result = query.executeQuery()) {
if (!result.next()) throw new StoreException(404, "NoSuchBucket", "Bucket not found");
return VersioningState.valueOf(result.getString(1));
}
}
}
private static VersioningState readVersioningState(Connection connection, String bucket) throws SQLException {
try (PreparedStatement query = connection.prepareStatement(
"SELECT versioning_state FROM cluster_buckets WHERE name=?")) {
query.setString(1, bucket);
try (ResultSet result = query.executeQuery()) {
if (!result.next()) throw new StoreException(404, "NoSuchBucket", "Bucket not found");
return VersioningState.valueOf(result.getString(1));
}
}
}
@Override public void createBucket(String name) throws IOException {
if (!name.matches("[a-z0-9][a-z0-9-]{1,61}[a-z0-9]"))
throw new StoreException(400, "InvalidBucketName", "Invalid bucket name");
try (Connection connection = connect()) {
connection.setAutoCommit(false);
try {
try (var statement = connection.createStatement()) {
statement.execute("SELECT pg_advisory_xact_lock(6834071092784)");
}
try (var statement = connection.createStatement();
ResultSet count = statement.executeQuery("SELECT count(*) FROM cluster_buckets")) {
count.next();
if (count.getLong(1) >= 1000)
throw new StoreException(400, "TooManyBuckets", "Bucket limit reached");
}
try (PreparedStatement insert = connection.prepareStatement(
"INSERT INTO cluster_buckets (name, created_at) VALUES (?, ?) ON CONFLICT DO NOTHING")) {
insert.setString(1, name);
insert.setLong(2, Instant.now().toEpochMilli());
if (insert.executeUpdate() == 0)
throw new StoreException(409, "BucketAlreadyOwnedByYou", "Bucket already exists");
}
try (PreparedStatement insert = connection.prepareStatement(
"INSERT INTO cluster_usage VALUES (?, 0)")) {
insert.setString(1, name);
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); }
}
@Override public void deleteBucket(String name) throws IOException {
try (Connection connection = connect()) {
connection.setAutoCommit(false);
try {
lockUsage(connection, name);
try (PreparedStatement check = connection.prepareStatement(
"SELECT EXISTS (SELECT 1 FROM cluster_object_versions WHERE bucket=?) OR EXISTS " +
"(SELECT 1 FROM cluster_uploads WHERE bucket=?)")) {
check.setString(1, name);
check.setString(2, name);
try (ResultSet result = check.executeQuery()) {
result.next();
if (result.getBoolean(1))
throw new StoreException(409, "BucketNotEmpty", "Bucket contains objects or uploads");
}
}
try (PreparedStatement delete = connection.prepareStatement("DELETE FROM cluster_buckets WHERE name=?")) {
delete.setString(1, name);
if (delete.executeUpdate() == 0)
throw new StoreException(404, "NoSuchBucket", "Bucket not found");
}
try (PreparedStatement delete = connection.prepareStatement("DELETE FROM cluster_usage WHERE bucket=?")) {
delete.setString(1, name);
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); }
}
private static void lockGc(Connection connection, boolean shared) throws SQLException {
try (var statement = connection.createStatement()) {
statement.execute("SELECT pg_advisory_lock" + (shared ? "_shared" : "") + "(6834071092783)");
@@ -66,26 +230,33 @@ final class ClusterStore implements ObjectStorage, MultipartStorage {
}
@Override public Metadata put(String bucket, String key, InputStream input, long length, String expectedHash,
String checksum, boolean createOnly, String contentType) throws IOException {
String checksum, boolean createOnly, String contentType,
Map<String, String> userMetadata, Map<String, String> tags,
java.util.function.Supplier<Map<String, String>> checksums,
Map<String, String> acl) throws IOException {
validatePut(bucket, length, contentType);
MessageDigest md5 = digest("MD5");
Crc64Nvme crc64 = new Crc64Nvme();
Path staged = Files.createTempFile("objectstore-cluster-", ".pending");
try {
byte[] fullHash = stageInput(staged, input, length, expectedHash, checksum, md5);
byte[] fullHash = stageInput(staged, input, length, expectedHash, checksum, md5, crc64);
Map<String, String> suppliedChecksums = checksums.get();
Map<String, String> storedChecksums = suppliedChecksums.isEmpty() ?
Map.of("x-amz-checksum-crc64nvme", crc64.encoded()) : Map.copyOf(suppliedChecksums);
checkCapacity(bucket, key, length, createOnly);
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(connection, metadata, segments, createOnly);
return metadata;
HexFormat.of().formatHex(md5.digest()), fullHash, bucket, key, contentType,
Map.copyOf(userMetadata), Map.copyOf(tags), null, storedChecksums, Map.copyOf(acl));
return persistObject(connection, metadata, segments, createOnly);
} catch (SQLException error) { throw databaseError(error); }
} finally { Files.deleteIfExists(staged); }
}
private void validatePut(String bucket, long length, String contentType) {
if (!configuredBucket.equals(bucket)) throw new StoreException(404, "NoSuchBucket", "Bucket not found");
private void validatePut(String bucket, long length, String contentType) throws IOException {
bucket(bucket);
if (length < 0) throw new StoreException(411, "MissingContentLength", "Content-Length is required");
if (length > maxObject) throw new StoreException(413, "EntityTooLarge", "Object exceeds the configured size limit");
if (!nodes.availableHostsAtLeast(2, testNodeDomains))
@@ -95,7 +266,7 @@ final class ClusterStore implements ObjectStorage, MultipartStorage {
}
private byte[] stageInput(Path staged, InputStream input, long length, String expectedHash,
String checksum, MessageDigest md5) throws IOException {
String checksum, MessageDigest md5, Crc64Nvme crc64) throws IOException {
MessageDigest sha = digest("SHA-256");
try (OutputStream output = Files.newOutputStream(staged)) {
byte[] buffer = new byte[65536];
@@ -106,13 +277,14 @@ final class ClusterStore implements ObjectStorage, MultipartStorage {
if (count == 0) continue;
sha.update(buffer, 0, count);
md5.update(buffer, 0, count);
crc64.update(buffer, 0, count);
output.write(buffer, 0, count);
remaining -= count;
}
}
if (input.read() != -1) throw new StoreException(413, "EntityTooLarge", "Payload exceeds declared size");
byte[] fullHash = sha.digest();
if (!HexFormat.of().formatHex(fullHash).equals(expectedHash))
if (expectedHash != null && !HexFormat.of().formatHex(fullHash).equals(expectedHash))
throw new StoreException(400, "XAmzContentSHA256Mismatch", "Payload hash mismatch");
if (checksum != null && !Base64.getEncoder().encodeToString(fullHash).equals(checksum))
throw new StoreException(400, "BadDigest", "SHA-256 checksum mismatch");
@@ -121,18 +293,14 @@ final class ClusterStore implements ObjectStorage, MultipartStorage {
private void checkCapacity(String bucket, String key, long length, boolean createOnly) throws IOException {
try (Connection connection = connect()) {
bucket(bucket);
long previous = currentLength(connection, bucket, key);
if (createOnly && previous >= 0)
throw new StoreException(412, "PreconditionFailed", "Object already exists");
try (PreparedStatement query = connection.prepareStatement("SELECT used_bytes FROM cluster_usage WHERE bucket=?")) {
query.setString(1, bucket);
try (ResultSet result = query.executeQuery()) {
if (!result.next()) throw new SQLException("Bucket quota row is missing");
if (maxTotal - (result.getLong(1) - Math.max(0, previous)) -
stagedBytes(connection, bucket) < length)
throw new StoreException(507, "InsufficientStorage", "Store capacity limit reached");
}
}
long replaced = readVersioningState(connection, bucket) == VersioningState.ENABLED ? 0 :
nullVersionLength(connection, bucket, key);
if (maxTotal - (occupiedBytes(connection) - replaced) < length)
throw new StoreException(507, "InsufficientStorage", "Store capacity limit reached");
} catch (SQLException error) { throw databaseError(error); }
}
@@ -170,8 +338,8 @@ final class ClusterStore implements ObjectStorage, MultipartStorage {
return segments;
}
private void persistObject(Connection connection, Metadata metadata, List<Segment> segments,
boolean createOnly) throws IOException {
private Metadata 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();
@@ -179,11 +347,12 @@ final class ClusterStore implements ObjectStorage, MultipartStorage {
connection.setAutoCommit(false);
try {
long used = lockUsage(connection, bucket);
VersioningState state = versioningState(connection, bucket);
long previous = currentLength(connection, bucket, key);
if (createOnly && previous >= 0)
throw new StoreException(412, "PreconditionFailed", "Object already exists");
if (maxTotal - (used - Math.max(0, previous)) -
stagedBytes(connection, bucket) < length)
long replaced = state == VersioningState.ENABLED ? 0 : nullVersionLength(connection, bucket, key);
if (maxTotal - (occupiedBytes(connection) - replaced) < 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', ?)")) {
@@ -199,22 +368,9 @@ final class ClusterStore implements ObjectStorage, MultipartStorage {
}
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) + 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();
}
Metadata stored = publishObject(connection, metadata, generation, used, replaced, state);
connection.commit();
return stored;
} catch (SQLException | RuntimeException error) {
connection.rollback();
if (error instanceof SQLException sql) throw databaseError(sql);
@@ -223,8 +379,96 @@ final class ClusterStore implements ObjectStorage, MultipartStorage {
} 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");
private static long nullVersionLength(Connection connection, String bucket, String key) throws SQLException {
try (PreparedStatement query = connection.prepareStatement(
"SELECT length FROM cluster_object_versions WHERE bucket=? AND object_key=? AND version_id='null'")) {
query.setString(1, bucket);
query.setString(2, key);
try (ResultSet result = query.executeQuery()) {
return result.next() ? result.getLong(1) : 0;
}
}
}
private Metadata publishObject(Connection connection, Metadata metadata, UUID generation,
long used, long replaced, VersioningState state) throws SQLException {
String bucket = metadata.bucket(), key = metadata.key();
String id = state == VersioningState.ENABLED ? UUID.randomUUID().toString() : "null";
if (id.equals("null")) {
try (PreparedStatement delete = connection.prepareStatement(
"DELETE FROM cluster_object_versions WHERE bucket=? AND object_key=? AND version_id='null'")) {
delete.setString(1, bucket);
delete.setString(2, key);
delete.executeUpdate();
}
}
try (PreparedStatement insert = connection.prepareStatement(
"INSERT INTO cluster_object_versions (bucket, object_key, version_id, delete_marker, generation, " +
"length, modified, etag, sha256, content_type, user_metadata, tags, checksum_metadata, acl) " +
"VALUES (?, ?, ?, false, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)")) {
insert.setString(1, bucket);
insert.setString(2, key);
insert.setString(3, id);
insert.setObject(4, generation);
insert.setLong(5, metadata.length());
insert.setLong(6, metadata.modified());
insert.setString(7, metadata.etag());
insert.setBytes(8, metadata.sha256());
insert.setString(9, metadata.contentType());
insert.setBytes(10, ObjectAttributes.encode(metadata.userMetadata(), 4096));
insert.setBytes(11, ObjectAttributes.encode(metadata.tags(), 8192));
insert.setBytes(12, ObjectAttributes.encode(metadata.checksums(), 512));
insert.setBytes(13, ObjectAttributes.encode(metadata.acl(), 2048));
insert.executeUpdate();
}
setHead(connection, bucket, key, id);
writeCurrentObject(connection, metadata, generation);
try (PreparedStatement update = connection.prepareStatement(
"UPDATE cluster_usage SET used_bytes=? WHERE bucket=?")) {
update.setLong(1, used - replaced + metadata.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();
}
return new Metadata(metadata.length(), metadata.modified(), metadata.etag(), metadata.sha256(),
bucket, key, metadata.contentType(), metadata.userMetadata(), metadata.tags(),
state == VersioningState.NEVER ? null : id, metadata.checksums(), metadata.acl());
}
private static void setHead(Connection connection, String bucket, String key, String id) throws SQLException {
try (PreparedStatement update = connection.prepareStatement(
"INSERT INTO cluster_object_heads VALUES (?, ?, ?) ON CONFLICT (bucket, object_key) " +
"DO UPDATE SET version_id=EXCLUDED.version_id")) {
update.setString(1, bucket);
update.setString(2, key);
update.setString(3, id);
update.executeUpdate();
}
}
private static void writeCurrentObject(Connection connection, Metadata metadata,
UUID generation) throws SQLException {
try (PreparedStatement update = connection.prepareStatement(
"INSERT INTO cluster_objects (bucket, object_key, generation, length, modified, etag, sha256, " +
"content_type, user_metadata, tags) 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, " +
"user_metadata=EXCLUDED.user_metadata, tags=EXCLUDED.tags")) {
bindObject(update, metadata, generation);
update.executeUpdate();
}
}
@Override public String create(String bucket, String key, String contentType,
Map<String, String> userMetadata, Map<String, String> tags,
Map<String, String> acl) throws IOException {
bucket(bucket);
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();
@@ -241,12 +485,15 @@ final class ClusterStore implements ObjectStorage, MultipartStorage {
}
}
try (PreparedStatement insert = connection.prepareStatement(
"INSERT INTO cluster_uploads VALUES (?, ?, ?, ?, ?)")) {
"INSERT INTO cluster_uploads (upload_id, bucket, object_key, content_type, created_at, user_metadata, tags, acl) VALUES (?, ?, ?, ?, ?, ?, ?, ?)")) {
insert.setObject(1, id);
insert.setString(2, bucket);
insert.setString(3, key);
insert.setString(4, contentType);
insert.setLong(5, Instant.now().toEpochMilli());
insert.setBytes(6, ObjectAttributes.encode(userMetadata, 4096));
insert.setBytes(7, ObjectAttributes.encode(tags, 8192));
insert.setBytes(8, ObjectAttributes.encode(acl, 2048));
insert.executeUpdate();
}
connection.commit();
@@ -269,7 +516,7 @@ final class ClusterStore implements ObjectStorage, MultipartStorage {
Path staged = Files.createTempFile("objectstore-part-", ".pending");
MessageDigest md5 = digest("MD5");
try {
stageInput(staged, input, length, expectedHash, checksum, md5);
stageInput(staged, input, length, expectedHash, checksum, md5, new Crc64Nvme());
try (Connection connection = connect()) {
lockGc(connection, true);
List<Segment> segments = uploadSegments(staged, length);
@@ -279,7 +526,7 @@ final class ClusterStore implements ObjectStorage, MultipartStorage {
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)
if (maxTotal - (occupiedBytes(connection) - 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")) {
@@ -330,11 +577,12 @@ final class ClusterStore implements ObjectStorage, MultipartStorage {
connection.setAutoCommit(false);
try {
long used = lockUsage(connection, bucket);
VersioningState state = versioningState(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");
Crc64Nvme crc64 = new Crc64Nvme();
List<Segment> selected = new ArrayList<>();
long total = 0;
int last = 0;
@@ -353,6 +601,7 @@ final class ClusterStore implements ObjectStorage, MultipartStorage {
byte[] bytes = readableSegment(segment);
if (bytes == null) throw new StoreException(503, "SlowDown", "A part has no verified replica");
fullHash.update(bytes);
crc64.update(bytes, 0, bytes.length);
partHash.update(bytes);
selected.add(segment);
}
@@ -361,12 +610,13 @@ final class ClusterStore implements ObjectStorage, MultipartStorage {
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)
long replaced = state == VersioningState.ENABLED ? 0 : nullVersionLength(connection, bucket, key);
if (maxTotal - (occupiedBytes(connection) - replaced - 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());
bucket, key, upload.contentType(), upload.userMetadata(), upload.tags(), null,
Map.of("x-amz-checksum-crc64nvme", crc64.encoded()), upload.acl());
UUID generation = UUID.randomUUID();
try (PreparedStatement insert = connection.prepareStatement(
"INSERT INTO cluster_segments (generation, ordinal, segment_id, length, sha256, replicas, replica_ids) VALUES (?, ?, ?, ?, ?, 'v2', ?)")) {
@@ -382,27 +632,13 @@ final class ClusterStore implements ObjectStorage, MultipartStorage {
}
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();
}
Metadata stored = publishObject(connection, metadata, generation, used, replaced, state);
try (PreparedStatement delete = connection.prepareStatement("DELETE FROM cluster_uploads WHERE upload_id=?")) {
delete.setObject(1, uploadId);
delete.executeUpdate();
}
connection.commit();
return metadata;
return stored;
} catch (SQLException | IOException | RuntimeException error) {
connection.rollback();
if (error instanceof SQLException sql) throw databaseError(sql);
@@ -460,7 +696,7 @@ final class ClusterStore implements ObjectStorage, MultipartStorage {
}
@Override public List<UploadInfo> listUploads(String bucket, String prefix) throws IOException {
if (!configuredBucket.equals(bucket)) throw new StoreException(404, "NoSuchBucket", "Bucket not found");
bucket(bucket);
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=?")) {
@@ -494,6 +730,10 @@ final class ClusterStore implements ObjectStorage, MultipartStorage {
}
@Override public OpenObject open(String bucket, String key) throws IOException {
return open(bucket, key, null);
}
@Override public OpenObject open(String bucket, String key, String versionId) throws IOException {
try (Connection connection = connect()) {
connection.setAutoCommit(false);
connection.setTransactionIsolation(Connection.TRANSACTION_REPEATABLE_READ);
@@ -502,14 +742,32 @@ final class ClusterStore implements ObjectStorage, MultipartStorage {
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=?")) {
versionId == null ?
"SELECT v.version_id, v.delete_marker, v.generation, v.length, v.modified, v.etag, " +
"v.sha256, v.content_type, v.user_metadata, v.tags, v.checksum_metadata, v.acl FROM cluster_object_heads h " +
"JOIN cluster_object_versions v ON v.bucket=h.bucket AND v.object_key=h.object_key " +
"AND v.version_id=h.version_id WHERE h.bucket=? AND h.object_key=?" :
"SELECT version_id, delete_marker, generation, length, modified, etag, sha256, " +
"content_type, user_metadata, tags, checksum_metadata, acl FROM cluster_object_versions WHERE bucket=? " +
"AND object_key=? AND version_id=?")) {
query.setString(1, bucket);
query.setString(2, key);
if (versionId != null) query.setString(3, versionId);
try (ResultSet result = query.executeQuery()) {
if (!result.next()) throw new StoreException(404, "NoSuchKey", "Object not found");
generation = (UUID) result.getObject(1);
metadata = new Metadata(result.getLong(2), result.getLong(3), result.getString(4),
result.getBytes(5), bucket, key, result.getString(6));
if (!result.next()) throw new StoreException(404,
versionId == null ? "NoSuchKey" : "NoSuchVersion", "Object version not found");
if (result.getBoolean(2))
throw StoreException.deletedVersion(result.getString(1), result.getLong(5),
versionId != null);
generation = (UUID) result.getObject(3);
String storedId = result.getString(1);
boolean unversioned = versionId == null && storedId.equals("null") &&
readVersioningState(connection, bucket) == VersioningState.NEVER;
metadata = new Metadata(result.getLong(4), result.getLong(5), result.getString(6),
result.getBytes(7), bucket, key, result.getString(8),
ObjectAttributes.decode(result.getBytes(9)), ObjectAttributes.decode(result.getBytes(10)),
unversioned ? null : storedId, ObjectAttributes.decode(result.getBytes(11)),
ObjectAttributes.decode(result.getBytes(12)));
}
}
List<Segment> parts = new ArrayList<>();
@@ -539,30 +797,253 @@ final class ClusterStore implements ObjectStorage, MultipartStorage {
}
@Override public void delete(String bucket, String key) throws IOException {
delete(bucket, key, null);
}
@Override public DeleteResult delete(String bucket, String key, String versionId) throws IOException {
try (Connection connection = connect()) {
connection.setAutoCommit(false);
try {
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();
VersioningState state = versioningState(connection, bucket);
long removedLength = 0;
boolean removedMarker = false;
String resultId = null;
if (versionId != null) {
try (PreparedStatement query = connection.prepareStatement(
"SELECT delete_marker, length FROM cluster_object_versions WHERE bucket=? " +
"AND object_key=? AND version_id=?")) {
query.setString(1, bucket);
query.setString(2, key);
query.setString(3, versionId);
try (ResultSet result = query.executeQuery()) {
if (!result.next())
throw new StoreException(404, "NoSuchVersion", "Object version not found");
removedMarker = result.getBoolean(1);
removedLength = removedMarker ? 0 : result.getLong(2);
}
}
removeVersionRow(connection, bucket, key, versionId);
refreshCurrent(connection, bucket, key);
resultId = versionId;
} else if (state == VersioningState.NEVER) {
removedLength = nullVersionLength(connection, bucket, key);
removeVersionRow(connection, bucket, key, "null");
refreshCurrent(connection, bucket, key);
} else {
resultId = state == VersioningState.ENABLED ? UUID.randomUUID().toString() : "null";
if (state == VersioningState.SUSPENDED) {
removedLength = nullVersionLength(connection, bucket, key);
removeVersionRow(connection, bucket, key, "null");
}
try (PreparedStatement insert = connection.prepareStatement(
"INSERT INTO cluster_object_versions (bucket, object_key, version_id, " +
"delete_marker, modified) VALUES (?, ?, ?, true, ?)")) {
insert.setString(1, bucket);
insert.setString(2, key);
insert.setString(3, resultId);
insert.setLong(4, Instant.now().toEpochMilli());
insert.executeUpdate();
}
setHead(connection, bucket, key, resultId);
removeCurrentObject(connection, bucket, key);
recordTombstone(connection, bucket, key);
}
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 cluster_usage SET used_bytes=? WHERE bucket=?")) {
update.setLong(1, used - removedLength);
update.setString(2, bucket);
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();
connection.commit();
return new DeleteResult(resultId, versionId == null && state != VersioningState.NEVER || removedMarker);
} 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); }
}
private static void removeVersionRow(Connection connection, String bucket, String key,
String versionId) throws SQLException {
try (PreparedStatement delete = connection.prepareStatement(
"DELETE FROM cluster_object_versions WHERE bucket=? AND object_key=? AND version_id=?")) {
delete.setString(1, bucket);
delete.setString(2, key);
delete.setString(3, versionId);
delete.executeUpdate();
}
}
private static void removeCurrentObject(Connection connection, String bucket, String key) throws SQLException {
try (PreparedStatement delete = connection.prepareStatement(
"DELETE FROM cluster_objects WHERE bucket=? AND object_key=?")) {
delete.setString(1, bucket);
delete.setString(2, key);
delete.executeUpdate();
}
}
private static void recordTombstone(Connection connection, String bucket, String key) throws SQLException {
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.executeUpdate();
}
}
private static void refreshCurrent(Connection connection, String bucket, String key)
throws SQLException, IOException {
try (PreparedStatement latest = connection.prepareStatement(
"SELECT version_id, delete_marker, generation, length, modified, etag, sha256, " +
"content_type, user_metadata, tags FROM cluster_object_versions WHERE bucket=? " +
"AND object_key=? ORDER BY sequence DESC LIMIT 1")) {
latest.setString(1, bucket);
latest.setString(2, key);
try (ResultSet result = latest.executeQuery()) {
if (!result.next()) {
try (PreparedStatement delete = connection.prepareStatement(
"DELETE FROM cluster_object_heads WHERE bucket=? AND object_key=?")) {
delete.setString(1, bucket);
delete.setString(2, key);
delete.executeUpdate();
}
removeCurrentObject(connection, bucket, key);
recordTombstone(connection, bucket, key);
return;
}
setHead(connection, bucket, key, result.getString(1));
if (result.getBoolean(2)) {
removeCurrentObject(connection, bucket, key);
recordTombstone(connection, bucket, key);
} else {
Metadata metadata = new Metadata(result.getLong(4), result.getLong(5),
result.getString(6), result.getBytes(7), bucket, key, result.getString(8),
ObjectAttributes.decode(result.getBytes(9)), ObjectAttributes.decode(result.getBytes(10)));
writeCurrentObject(connection, metadata, (UUID) result.getObject(3));
try (PreparedStatement delete = connection.prepareStatement(
"DELETE FROM cluster_tombstones WHERE bucket=? AND object_key=?")) {
delete.setString(1, bucket);
delete.setString(2, key);
delete.executeUpdate();
}
}
}
}
}
@Override public Map<String, String> tags(String bucket, String key) throws IOException {
return tags(bucket, key, null);
}
@Override public Map<String, String> tags(String bucket, String key, String versionId) throws IOException {
try (Connection connection = connect(); PreparedStatement query = connection.prepareStatement(
versionId == null ? "SELECT v.tags, v.delete_marker FROM cluster_object_heads h " +
"JOIN cluster_object_versions v ON v.bucket=h.bucket AND v.object_key=h.object_key " +
"AND v.version_id=h.version_id WHERE h.bucket=? AND h.object_key=?" :
"SELECT tags, delete_marker FROM cluster_object_versions WHERE bucket=? AND object_key=? " +
"AND version_id=?")) {
query.setString(1, bucket);
query.setString(2, key);
if (versionId != null) query.setString(3, versionId);
try (ResultSet result = query.executeQuery()) {
if (!result.next() || result.getBoolean(2))
throw new StoreException(404, versionId == null ? "NoSuchKey" : "NoSuchVersion",
"Object version not found");
return ObjectAttributes.decode(result.getBytes(1));
}
} catch (SQLException error) { throw databaseError(error); }
}
@Override public void setTags(String bucket, String key, Map<String, String> tags) throws IOException {
setTags(bucket, key, null, tags);
}
@Override public void setTags(String bucket, String key, String versionId,
Map<String, String> tags) throws IOException {
byte[] encoded = ObjectAttributes.encode(tags, 8192);
try (Connection connection = connect()) {
connection.setAutoCommit(false);
try {
lockUsage(connection, bucket);
String selected = versionId;
if (selected == null) {
try (PreparedStatement head = connection.prepareStatement(
"SELECT version_id FROM cluster_object_heads WHERE bucket=? AND object_key=?")) {
head.setString(1, bucket);
head.setString(2, key);
try (ResultSet result = head.executeQuery()) {
if (!result.next()) throw new StoreException(404, "NoSuchKey", "Object not found");
selected = result.getString(1);
}
}
}
try (PreparedStatement update = connection.prepareStatement(
"UPDATE cluster_object_versions SET tags=? WHERE bucket=? AND object_key=? " +
"AND version_id=? AND NOT delete_marker")) {
update.setBytes(1, encoded);
update.setString(2, bucket);
update.setString(3, key);
update.setString(4, selected);
if (update.executeUpdate() == 0)
throw new StoreException(404, versionId == null ? "NoSuchKey" : "NoSuchVersion",
"Object version not found");
}
try (PreparedStatement update = connection.prepareStatement(
"UPDATE cluster_objects SET tags=? WHERE bucket=? AND object_key=? AND EXISTS " +
"(SELECT 1 FROM cluster_object_heads WHERE bucket=? AND object_key=? AND version_id=?)")) {
update.setBytes(1, encoded);
update.setString(2, bucket);
update.setString(3, key);
update.setString(4, bucket);
update.setString(5, key);
update.setString(6, selected);
update.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 void setObjectAcl(String bucket, String key, String versionId,
Map<String, String> acl) throws IOException {
byte[] encoded = ObjectAttributes.encode(acl, 2048);
try (Connection connection = connect()) {
connection.setAutoCommit(false);
try {
lockUsage(connection, bucket);
String selected = versionId;
if (selected == null) {
try (PreparedStatement head = connection.prepareStatement(
"SELECT version_id FROM cluster_object_heads WHERE bucket=? AND object_key=?")) {
head.setString(1, bucket);
head.setString(2, key);
try (ResultSet result = head.executeQuery()) {
if (!result.next()) throw new StoreException(404, "NoSuchKey", "Object not found");
selected = result.getString(1);
}
}
}
try (PreparedStatement update = connection.prepareStatement(
"UPDATE cluster_object_versions SET acl=? WHERE bucket=? AND object_key=? " +
"AND version_id=? AND NOT delete_marker")) {
update.setBytes(1, encoded);
update.setString(2, bucket);
update.setString(3, key);
update.setString(4, selected);
if (update.executeUpdate() == 0)
throw new StoreException(404, versionId == null ? "NoSuchKey" : "NoSuchVersion",
"Object version not found");
}
connection.commit();
} catch (SQLException | RuntimeException error) {
@@ -592,6 +1073,61 @@ final class ClusterStore implements ObjectStorage, MultipartStorage {
} catch (SQLException error) { throw databaseError(error); }
}
@Override public VersionPage listVersions(String bucket, String prefix, String keyMarker,
String versionMarker, int maxKeys) throws IOException {
if (versionMarker != null && keyMarker == null)
throw new StoreException(400, "InvalidArgument", "Version marker requires a key marker");
bucket(bucket);
if (maxKeys == 0) return new VersionPage(List.of(), null, null, false);
List<VersionEntry> page = new ArrayList<>();
String nextKey = null, nextVersion = null;
boolean truncated = false;
try (Connection connection = connect()) {
connection.setAutoCommit(false);
try (PreparedStatement query = connection.prepareStatement(
"SELECT v.object_key, v.version_id, v.delete_marker, v.length, v.modified, v.etag, " +
"v.sha256, v.content_type, v.user_metadata, v.tags, h.version_id=v.version_id, " +
"v.checksum_metadata " +
"FROM cluster_object_versions v LEFT JOIN cluster_object_heads h ON " +
"h.bucket=v.bucket AND h.object_key=v.object_key WHERE v.bucket=? AND v.object_key>=? " +
"ORDER BY v.object_key, v.sequence DESC")) {
query.setString(1, bucket);
query.setString(2, keyMarker != null && keyMarker.compareTo(prefix) > 0 ? keyMarker : prefix);
query.setFetchSize(128);
try (ResultSet rows = query.executeQuery()) {
boolean pastMarker = versionMarker == null;
while (rows.next()) {
String key = rows.getString(1), id = rows.getString(2);
if (!key.startsWith(prefix)) break;
if (keyMarker != null && key.compareTo(keyMarker) < 0) continue;
if (keyMarker != null && key.compareTo(keyMarker) > 0) pastMarker = true;
if (keyMarker != null && key.equals(keyMarker)) {
if (versionMarker == null) continue;
if (!pastMarker) {
if (id.equals(versionMarker)) pastMarker = true;
continue;
}
}
if (page.size() == maxKeys) {
truncated = true;
break;
}
boolean marker = rows.getBoolean(3);
Metadata metadata = marker ? null : new Metadata(rows.getLong(4), rows.getLong(5),
rows.getString(6), rows.getBytes(7), bucket, key, rows.getString(8),
ObjectAttributes.decode(rows.getBytes(9)), ObjectAttributes.decode(rows.getBytes(10)),
id, ObjectAttributes.decode(rows.getBytes(12)));
page.add(new VersionEntry(key, id, rows.getLong(5), marker, rows.getBoolean(11), metadata));
nextKey = key;
nextVersion = id;
}
}
}
connection.commit();
} catch (SQLException error) { throw databaseError(error); }
return new VersionPage(page, truncated ? nextKey : null, truncated ? nextVersion : null, truncated);
}
private static ListPage readListPage(ResultSet result, String bucket, String prefix, String delimiter,
int maxKeys, String after) throws SQLException {
List<ListedObject> entries = new ArrayList<>();
@@ -631,15 +1167,27 @@ final class ClusterStore implements ObjectStorage, MultipartStorage {
}
private long lockUsage(Connection connection, String bucket) throws SQLException {
try (var statement = connection.createStatement()) {
statement.execute("SELECT pg_advisory_xact_lock(6834071092784)");
}
try (PreparedStatement query = connection.prepareStatement("SELECT used_bytes FROM cluster_usage WHERE bucket=? FOR UPDATE")) {
query.setString(1, bucket);
try (ResultSet result = query.executeQuery()) {
if (!result.next()) throw new SQLException("Bucket quota row is missing");
if (!result.next()) throw new StoreException(404, "NoSuchBucket", "Bucket not found");
return result.getLong(1);
}
}
}
private static long occupiedBytes(Connection connection) throws SQLException {
try (var statement = connection.createStatement(); ResultSet result = statement.executeQuery(
"SELECT (SELECT COALESCE(sum(used_bytes), 0) FROM cluster_usage) + " +
"(SELECT COALESCE(sum(length), 0) FROM cluster_upload_parts)")) {
result.next();
return result.getLong(1);
}
}
private static UUID uploadId(String id) {
try {
if (id == null || !id.matches("[0-9a-f-]{36}")) throw new IllegalArgumentException();
@@ -650,15 +1198,16 @@ final class ClusterStore implements ObjectStorage, MultipartStorage {
}
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=?" +
throws SQLException, IOException {
String sql = "SELECT bucket, object_key, content_type, created_at, user_metadata, tags, acl 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));
return new Upload(result.getString(3), ObjectAttributes.decode(result.getBytes(5)),
ObjectAttributes.decode(result.getBytes(6)), ObjectAttributes.decode(result.getBytes(7)));
}
}
}
@@ -755,6 +1304,8 @@ final class ClusterStore implements ObjectStorage, MultipartStorage {
update.setString(6, data.etag());
update.setBytes(7, data.sha256());
update.setString(8, data.contentType());
update.setBytes(9, ObjectAttributes.encode(data.userMetadata(), 4096));
update.setBytes(10, ObjectAttributes.encode(data.tags(), 8192));
}
private static List<UUID> replicaIds(ResultSet result, int column) throws SQLException, IOException {
java.sql.Array value = result.getArray(column);
@@ -809,7 +1360,7 @@ final class ClusterStore implements ObjectStorage, MultipartStorage {
}
try (PreparedStatement query = reader.prepareStatement(
"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 " +
"FROM cluster_segments s JOIN cluster_object_versions 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")) {
@@ -924,7 +1475,7 @@ final class ClusterStore implements ObjectStorage, MultipartStorage {
}
lockGc(connection, false);
try (PreparedStatement referenced = connection.prepareStatement(
"SELECT EXISTS (SELECT 1 FROM cluster_segments s JOIN cluster_objects o " +
"SELECT EXISTS (SELECT 1 FROM cluster_segments s JOIN cluster_object_versions 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))");
+39
View File
@@ -0,0 +1,39 @@
package cloud.lunarsky.store;
import java.util.zip.Checksum;
import java.util.Base64;
import java.nio.ByteBuffer;
final class Crc64Nvme implements Checksum {
private static final long POLYNOMIAL = 0x9a6c9329ac4bc9b5L;
private static final long[] TABLE = table();
private long state = -1L;
private static long[] table() {
long[] values = new long[256];
for (int index = 0; index < values.length; index++) {
long value = index;
for (int bit = 0; bit < 8; bit++)
value = (value >>> 1) ^ ((value & 1L) == 0 ? 0 : POLYNOMIAL);
values[index] = value;
}
return values;
}
@Override public void update(int value) {
state = (state >>> 8) ^ TABLE[(int) (state ^ value) & 255];
}
@Override public void update(byte[] bytes, int offset, int length) {
java.util.Objects.checkFromIndexSize(offset, length, bytes.length);
for (int i = offset; i < offset + length; i++) update(bytes[i]);
}
@Override public long getValue() { return ~state; }
String encoded() {
return Base64.getEncoder().encodeToString(ByteBuffer.allocate(8).putLong(getValue()).array());
}
@Override public void reset() { state = -1L; }
}
+635 -17
View File
@@ -19,23 +19,40 @@ import cloud.lunarsky.store.ObjectStorage.ListPage;
final class DiskStore implements ObjectStorage {
private static final long MAGIC_V1 = 0x4c534f424a303031L;
private static final long MAGIC_V2 = 0x4c534f424a303032L;
private static final long MAGIC_V3 = 0x4c534f424a303033L;
private static final long MAGIC_V4 = 0x4c534f424a303034L;
private static final long MAGIC_V5 = 0x4c534f424a303035L;
private static final int HEADER_V1 = 72;
private static final int HEADER_V2 = 78;
private final Path root, objects, temporary;
private static final int HEADER_V3 = 82;
private static final int CHECKSUM_AREA = 512;
private static final int HEADER_V4 = HEADER_V3 + 2 + CHECKSUM_AREA;
private static final int ACL_AREA = 2048;
private static final int HEADER_V5 = HEADER_V4 + 2;
private static final int BUCKET_MAGIC = 0x4c534243;
private static final int BUCKET_MAGIC_V2 = 0x4c534244;
private static final int BUCKET_MAGIC_V3 = 0x4c534245;
private static final int VERSION_MAGIC = 0x4c53564d;
private final Path root, objects, temporary, catalog, versions;
private final FileChannel lockChannel;
private final FileLock processLock;
private final long maxObject, maxTotal;
private final Object[] locks = new Object[128];
private final NavigableMap<String, Metadata> index = new TreeMap<>();
private final NavigableMap<String, Bucket> buckets = new TreeMap<>();
private final NavigableMap<String, List<VersionRecord>> histories = new TreeMap<>();
private long used;
private long objectCount, legacyCount;
record Record(Metadata metadata, int headerLength) {}
private record VersionRecord(String id, String storageId, boolean marker, long modified) {}
DiskStore(Path root, long maxObject, long maxTotal) throws IOException {
this.root = root;
objects = root.resolve("objects");
temporary = root.resolve("pending");
catalog = root.resolve("buckets.bin");
versions = root.resolve("versions");
this.maxObject = maxObject;
this.maxTotal = maxTotal;
Arrays.setAll(locks, i -> new Object());
@@ -49,6 +66,7 @@ final class DiskStore implements ObjectStorage {
if (acquired == null) throw new IOException("Data directory is already in use");
Files.createDirectories(objects);
Files.createDirectories(temporary);
Files.createDirectories(versions);
syncDirectory(root);
try (var paths = Files.list(temporary)) {
for (Path p : paths.toList()) if (p.getFileName().toString().endsWith(".part")) Files.delete(p);
@@ -70,6 +88,37 @@ final class DiskStore implements ObjectStorage {
used = Math.addExact(used, meta.length());
}
}
if (Files.exists(catalog)) {
try (DataInputStream input = new DataInputStream(Files.newInputStream(catalog))) {
int magic = input.readInt();
if (magic != BUCKET_MAGIC && magic != BUCKET_MAGIC_V2 && magic != BUCKET_MAGIC_V3)
throw new IOException("Invalid bucket catalog");
int count = input.readInt();
if (count < 0 || count > 1000) throw new IOException("Invalid bucket catalog");
for (int i = 0; i < count; i++) {
String name = input.readUTF();
long created = input.readLong();
VersioningState state = VersioningState.NEVER;
if (magic == BUCKET_MAGIC_V2 || magic == BUCKET_MAGIC_V3) {
int ordinal = input.readUnsignedByte();
if (ordinal >= VersioningState.values().length)
throw new IOException("Invalid bucket versioning state");
state = VersioningState.values()[ordinal];
}
Map<String, String> acl = Map.of();
if (magic == BUCKET_MAGIC_V3) {
int size = input.readUnsignedShort();
if (size > ACL_AREA) throw new IOException("Invalid bucket ACL");
acl = ObjectAttributes.decode(input.readNBytes(size));
}
if (!validBucket(name) || created < 0 ||
buckets.put(name, new Bucket(name, created, state, acl)) != null)
throw new IOException("Invalid bucket catalog");
}
if (input.read() != -1) throw new IOException("Invalid bucket catalog");
}
}
loadHistories();
ready = true;
} finally {
if (!ready) {
@@ -88,12 +137,251 @@ final class DiskStore implements ObjectStorage {
Path root() { return root; }
long maxObject() { return maxObject; }
long maxTotal() { return maxTotal; }
@Override public Limits limits() { return new Limits(maxObject, maxTotal); }
synchronized long usedBytes() { return used; }
synchronized int indexedObjects() { return index.size(); }
synchronized long objectCount() { return objectCount; }
synchronized long legacyObjects() { return legacyCount; }
private static boolean validBucket(String name) {
return name.matches("[a-z0-9][a-z0-9-]{1,61}[a-z0-9]");
}
@Override public synchronized void ensureBucket(String bucket) throws IOException {
if (!buckets.containsKey(bucket)) createBucket(bucket);
}
@Override public synchronized Bucket bucket(String name) {
Bucket found = buckets.get(name);
if (found == null) throw new StoreException(404, "NoSuchBucket", "Bucket not found");
return found;
}
@Override public synchronized List<Bucket> buckets() { return List.copyOf(buckets.values()); }
@Override public synchronized void createBucket(String name) throws IOException {
if (!validBucket(name)) throw new StoreException(400, "InvalidBucketName", "Invalid bucket name");
if (buckets.containsKey(name))
throw new StoreException(409, "BucketAlreadyOwnedByYou", "Bucket already exists");
if (buckets.size() >= 1000) throw new StoreException(400, "TooManyBuckets", "Bucket limit reached");
NavigableMap<String, Bucket> next = new TreeMap<>(buckets);
next.put(name, new Bucket(name, Instant.now().toEpochMilli()));
saveBuckets(next);
buckets.clear();
buckets.putAll(next);
}
@Override public synchronized void deleteBucket(String name) throws IOException {
bucket(name);
if (index.values().stream().anyMatch(meta -> name.equals(meta.bucket())))
throw new StoreException(409, "BucketNotEmpty", "Bucket contains objects");
if (histories.entrySet().stream().anyMatch(entry -> entry.getKey().startsWith(name + "\0") &&
!entry.getValue().isEmpty()))
throw new StoreException(409, "BucketNotEmpty", "Bucket contains object versions");
if (legacyCount > 0) {
try (var paths = Files.walk(objects)) {
for (Path path : paths.filter(Files::isRegularFile).toList()) {
try (var input = new DataInputStream(Files.newInputStream(path))) {
Metadata metadata = readRecord(input).metadata();
if (metadata.key() == null && name.equals(metadata.bucket()))
throw new StoreException(409, "BucketNotEmpty", "Bucket contains legacy objects");
}
}
}
}
NavigableMap<String, Bucket> next = new TreeMap<>(buckets);
next.remove(name);
saveBuckets(next);
buckets.clear();
buckets.putAll(next);
for (String key : new ArrayList<>(histories.keySet())) {
if (!key.startsWith(name + "\0") || !histories.get(key).isEmpty()) continue;
String objectKey = key.substring(name.length() + 1);
Path directory = historyDirectory(name, objectKey);
Files.deleteIfExists(directory.resolve("manifest"));
syncDirectory(directory);
histories.remove(key);
}
}
private void saveBuckets(NavigableMap<String, Bucket> next) throws IOException {
Path pending = Files.createTempFile(temporary, "buckets-", ".part");
try {
try (DataOutputStream output = new DataOutputStream(Files.newOutputStream(pending))) {
output.writeInt(BUCKET_MAGIC_V3);
output.writeInt(next.size());
for (Bucket entry : next.values()) {
output.writeUTF(entry.name());
output.writeLong(entry.created());
output.writeByte(entry.versioning().ordinal());
byte[] acl = ObjectAttributes.encode(entry.acl(), ACL_AREA);
output.writeShort(acl.length);
output.write(acl);
}
}
try (FileChannel channel = FileChannel.open(pending, StandardOpenOption.WRITE)) { channel.force(true); }
Files.move(pending, catalog, StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING);
syncDirectory(root);
} finally { Files.deleteIfExists(pending); }
}
@Override public synchronized void setVersioning(String name, VersioningState state) throws IOException {
if (state == VersioningState.NEVER)
throw new StoreException(400, "InvalidArgument", "Versioning cannot be disabled after it is enabled");
Bucket old = bucket(name);
if (old.versioning() == VersioningState.NEVER && state == VersioningState.SUSPENDED)
throw new StoreException(400, "InvalidArgument", "Enable versioning before suspending it");
NavigableMap<String, Bucket> next = new TreeMap<>(buckets);
next.put(name, new Bucket(name, old.created(), state, old.acl()));
saveBuckets(next);
buckets.clear();
buckets.putAll(next);
}
@Override public synchronized void setBucketAcl(String name, Map<String, String> acl) throws IOException {
Bucket old = bucket(name);
NavigableMap<String, Bucket> next = new TreeMap<>(buckets);
next.put(name, new Bucket(name, old.created(), old.versioning(), Map.copyOf(acl)));
saveBuckets(next);
buckets.clear();
buckets.putAll(next);
}
private static String indexKey(String bucket, String key) { return bucket + "\0" + key; }
private Path historyDirectory(String bucket, String key) {
String id = SigV4.hex(SigV4.hash((bucket + "/" + key).getBytes(StandardCharsets.UTF_8)));
return versions.resolve(id.substring(0, 2)).resolve(id);
}
private Path versionPath(String bucket, String key, String storageId) {
return storageId.equals("legacy") ? objectPath(bucket, key) :
historyDirectory(bucket, key).resolve(storageId);
}
private static Metadata withVersion(Metadata old, String id, String bucket, String key) {
return new Metadata(old.length(), old.modified(), old.etag(), old.sha256(), bucket, key,
old.contentType(), old.userMetadata(), old.tags(), id, old.checksums(), old.acl());
}
private List<VersionRecord> history(String bucket, String key) throws IOException {
List<VersionRecord> found = histories.get(indexKey(bucket, key));
if (found != null) return found;
Metadata old = index.get(indexKey(bucket, key));
if (old == null && Files.isRegularFile(objectPath(bucket, key))) {
try (DataInputStream input = new DataInputStream(Files.newInputStream(objectPath(bucket, key)))) {
old = readRecord(input).metadata();
}
}
return old == null ? List.of() : List.of(new VersionRecord("null", "legacy", false, old.modified()));
}
private void saveHistory(String bucket, String key, List<VersionRecord> entries) throws IOException {
Path directory = historyDirectory(bucket, key);
if (!Files.isDirectory(directory)) {
Files.createDirectories(directory);
syncDirectory(directory.getParent());
}
Path pending = Files.createTempFile(temporary, "history-", ".part");
try {
try (DataOutputStream output = new DataOutputStream(Files.newOutputStream(pending))) {
output.writeInt(VERSION_MAGIC);
output.writeUTF(bucket);
output.writeUTF(key);
output.writeInt(entries.size());
for (VersionRecord entry : entries) {
output.writeUTF(entry.id());
output.writeUTF(entry.storageId());
output.writeBoolean(entry.marker());
output.writeLong(entry.modified());
}
}
try (FileChannel channel = FileChannel.open(pending, StandardOpenOption.WRITE)) {
channel.force(true);
}
Files.move(pending, directory.resolve("manifest"), StandardCopyOption.ATOMIC_MOVE,
StandardCopyOption.REPLACE_EXISTING);
syncDirectory(directory);
} finally { Files.deleteIfExists(pending); }
histories.put(indexKey(bucket, key), List.copyOf(entries));
}
private void loadHistories() throws IOException {
Set<Path> referenced = new HashSet<>();
try (var paths = Files.walk(versions)) {
for (Path manifest : paths.filter(p -> p.getFileName().toString().equals("manifest")).toList()) {
String bucket;
String key;
List<VersionRecord> entries = new ArrayList<>();
try (DataInputStream input = new DataInputStream(Files.newInputStream(manifest))) {
if (input.readInt() != VERSION_MAGIC) throw new IOException("Invalid version manifest: " + manifest);
bucket = input.readUTF();
key = input.readUTF();
int count = input.readInt();
if (count < 0 || count > 1_000_000 || !manifest.getParent().equals(historyDirectory(bucket, key)))
throw new IOException("Invalid version manifest: " + manifest);
Set<String> ids = new HashSet<>();
for (int i = 0; i < count; i++) {
String id = input.readUTF();
String storageId = input.readUTF();
boolean marker = input.readBoolean();
long modified = input.readLong();
if ((!id.equals("null") && !id.matches("[0-9a-f-]{36}")) ||
(!marker && !storageId.equals("legacy") && !storageId.matches("[0-9a-f-]{36}")) ||
(marker && !storageId.isEmpty()) || !ids.add(id) || modified < 0)
throw new IOException("Invalid version manifest entry: " + manifest);
entries.add(new VersionRecord(id, storageId, marker, modified));
if (!marker) {
Path file = versionPath(bucket, key, storageId);
if (!Files.isRegularFile(file)) throw new IOException("Missing object version: " + file);
referenced.add(file);
if (!storageId.equals("legacy")) {
Record record;
try (DataInputStream data = new DataInputStream(Files.newInputStream(file))) {
record = readRecord(data);
}
Metadata meta = record.metadata();
if (!bucket.equals(meta.bucket()) || !key.equals(meta.key()) ||
Files.size(file) - record.headerLength() != meta.length())
throw new IOException("Invalid object version: " + file);
objectCount++;
used = Math.addExact(used, meta.length());
}
}
}
if (input.read() != -1 || histories.put(indexKey(bucket, key), List.copyOf(entries)) != null)
throw new IOException("Invalid version manifest: " + manifest);
}
if (entries.isEmpty() || entries.getFirst().marker()) index.remove(indexKey(bucket, key));
else {
VersionRecord current = entries.getFirst();
Path file = versionPath(bucket, key, current.storageId());
try (DataInputStream input = new DataInputStream(Files.newInputStream(file))) {
index.put(indexKey(bucket, key),
withVersion(readRecord(input).metadata(), current.id(), bucket, key));
}
}
}
}
try (var paths = Files.walk(versions)) {
for (Path file : paths.filter(Files::isRegularFile).toList()) {
if (!file.getFileName().toString().equals("manifest") && !referenced.contains(file)) {
Files.delete(file);
syncDirectory(file.getParent());
}
}
}
for (var entry : histories.entrySet()) {
String[] parts = entry.getKey().split("\0", 2);
if (entry.getValue().stream().anyMatch(version -> version.storageId().equals("legacy") && !version.marker()))
continue;
Path orphan = objectPath(parts[0], parts[1]);
if (Files.isRegularFile(orphan)) {
try (DataInputStream input = new DataInputStream(Files.newInputStream(orphan))) {
Metadata old = readRecord(input).metadata();
used -= old.length();
if (old.key() == null) legacyCount--;
}
Files.delete(orphan);
objectCount--;
syncDirectory(orphan.getParent());
}
}
}
private Path objectPath(String bucket, String key) {
String id = SigV4.hex(SigV4.hash((bucket + "/" + key).getBytes(StandardCharsets.UTF_8)));
return objects.resolve(id.substring(0, 2)).resolve(id);
@@ -114,19 +402,24 @@ final class DiskStore implements ObjectStorage {
private Object lock(Path p) { return locks[(p.hashCode() & 0x7fffffff) % locks.length]; }
public Metadata put(String bucket, String key, InputStream input, long length, String expectedHash,
String checksum, boolean createOnly, String contentType) throws IOException {
String checksum, boolean createOnly, String contentType,
Map<String, String> userMetadata, Map<String, String> tags,
java.util.function.Supplier<Map<String, String>> checksums,
Map<String, String> acl) throws IOException {
if (length < 0) throw new StoreException(411, "MissingContentLength", "Content-Length is required");
if (length > maxObject) throw new StoreException(413, "EntityTooLarge", "Object exceeds the configured size limit");
byte[] bucketBytes = bucket.getBytes(StandardCharsets.UTF_8);
byte[] keyBytes = key.getBytes(StandardCharsets.UTF_8);
byte[] typeBytes = contentType.getBytes(StandardCharsets.UTF_8);
byte[] metadataBytes = ObjectAttributes.encode(userMetadata, 4096);
byte[] tagBytes = ObjectAttributes.encode(tags, 8192);
validateMetadataLengths(bucketBytes, keyBytes, typeBytes);
Path destination = object(bucket, key), pending = Files.createTempFile(temporary, "upload-", ".part");
try {
Metadata metadata = stagePut(pending, input, length, expectedHash, checksum,
bucket, key, contentType, bucketBytes, keyBytes, typeBytes);
installPending(destination, pending, metadata, createOnly);
return metadata;
bucket, key, contentType, bucketBytes, keyBytes, typeBytes,
metadataBytes, tagBytes, userMetadata, tags, checksums, acl);
return installPending(destination, pending, metadata, createOnly);
} finally { Files.deleteIfExists(pending); }
}
@@ -137,9 +430,17 @@ final class DiskStore implements ObjectStorage {
private Metadata stagePut(Path pending, InputStream input, long length, String expectedHash, String checksum,
String bucket, String key, String contentType,
byte[] bucketBytes, byte[] keyBytes, byte[] typeBytes) throws IOException {
int headerLength = HEADER_V2 + bucketBytes.length + keyBytes.length + typeBytes.length;
byte[] bucketBytes, byte[] keyBytes, byte[] typeBytes,
byte[] metadataBytes, byte[] tagBytes,
Map<String, String> userMetadata, Map<String, String> tags,
java.util.function.Supplier<Map<String, String>> checksums,
Map<String, String> acl) throws IOException {
byte[] aclBytes = ObjectAttributes.encode(acl, ACL_AREA);
int headerLength = HEADER_V5 + aclBytes.length +
bucketBytes.length + keyBytes.length + typeBytes.length +
metadataBytes.length + tagBytes.length;
MessageDigest sha = digest("SHA-256"), md5 = digest("MD5");
Crc64Nvme crc64 = new Crc64Nvme();
long count = 0;
try (OutputStream out = Files.newOutputStream(pending)) {
out.write(new byte[headerLength]);
@@ -151,30 +452,47 @@ final class DiskStore implements ObjectStorage {
throw new StoreException(413, "EntityTooLarge", "Payload exceeds declared size");
sha.update(buffer, 0, n);
md5.update(buffer, 0, n);
crc64.update(buffer, 0, n);
out.write(buffer, 0, n);
}
}
if (count != length) throw new StoreException(400, "IncompleteBody", "Payload length does not match Content-Length");
byte[] hash = sha.digest(), etag = md5.digest();
if (!MessageDigest.isEqual(hash, HexFormat.of().parseHex(expectedHash)))
if (expectedHash != null && !MessageDigest.isEqual(hash, HexFormat.of().parseHex(expectedHash)))
throw new StoreException(400, "XAmzContentSHA256Mismatch", "Payload hash mismatch");
if (checksum != null && !Base64.getEncoder().encodeToString(hash).equals(checksum))
throw new StoreException(400, "BadDigest", "SHA-256 checksum mismatch");
Map<String, String> suppliedChecksums = checksums.get();
Map<String, String> storedChecksums = suppliedChecksums.isEmpty() ?
Map.of("x-amz-checksum-crc64nvme", crc64.encoded()) : Map.copyOf(suppliedChecksums);
byte[] checksumBytes = ObjectAttributes.encode(storedChecksums, CHECKSUM_AREA);
long modified = Instant.now().toEpochMilli();
ByteBuffer header = ByteBuffer.allocate(headerLength).putLong(MAGIC_V2).putLong(count)
ByteBuffer header = ByteBuffer.allocate(headerLength).putLong(MAGIC_V5).putLong(count)
.putLong(modified).put(etag).put(hash).putShort((short) bucketBytes.length)
.putShort((short) keyBytes.length).putShort((short) typeBytes.length)
.put(bucketBytes).put(keyBytes).put(typeBytes);
.putShort((short) metadataBytes.length).putShort((short) tagBytes.length)
.putShort((short) checksumBytes.length);
header.position(header.position() + CHECKSUM_AREA - checksumBytes.length);
header.put(checksumBytes).putShort((short) aclBytes.length);
header.put(aclBytes).put(bucketBytes).put(keyBytes).put(typeBytes).put(metadataBytes).put(tagBytes);
header.flip();
try (FileChannel file = FileChannel.open(pending, StandardOpenOption.WRITE)) {
while (header.hasRemaining()) file.write(header, header.position());
file.force(true);
}
return new Metadata(count, modified, SigV4.hex(etag), hash, bucket, key, contentType);
return new Metadata(count, modified, SigV4.hex(etag), hash, bucket, key, contentType,
Map.copyOf(userMetadata), Map.copyOf(tags), null, storedChecksums, Map.copyOf(acl));
}
private void installPending(Path destination, Path pending, Metadata metadata, boolean createOnly) throws IOException {
private Metadata installPending(Path destination, Path pending, Metadata metadata,
boolean createOnly) throws IOException {
synchronized (lock(destination)) {
synchronized (this) {
Bucket configured = buckets.get(metadata.bucket());
if (configured != null && configured.versioning() != VersioningState.NEVER)
return installVersionedPending(destination, pending, metadata, createOnly,
configured.versioning());
}
long previous = 0;
boolean existed = Files.exists(destination);
boolean legacy = false;
@@ -187,6 +505,8 @@ final class DiskStore implements ObjectStorage {
}
}
synchronized (this) {
if (Files.exists(catalog) && !buckets.containsKey(metadata.bucket()))
throw new StoreException(404, "NoSuchBucket", "Bucket not found");
if (used - previous + metadata.length() > maxTotal)
throw new StoreException(507, "InsufficientStorage", "Store capacity limit reached");
Files.move(pending, destination, StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING);
@@ -196,16 +516,91 @@ final class DiskStore implements ObjectStorage {
index.put(indexKey(metadata.bucket(), metadata.key()), metadata);
syncDirectory(destination.getParent());
}
return metadata;
}
}
private Metadata installVersionedPending(Path destination, Path pending, Metadata metadata,
boolean createOnly, VersioningState state) throws IOException {
String bucket = metadata.bucket();
String key = metadata.key();
if (createOnly && index.containsKey(indexKey(bucket, key)))
throw new StoreException(412, "PreconditionFailed", "Object already exists");
List<VersionRecord> old = history(bucket, key);
String id = state == VersioningState.ENABLED ? UUID.randomUUID().toString() : "null";
String storageId = UUID.randomUUID().toString();
VersionRecord discarded = null;
long replaced = 0;
if (id.equals("null")) {
for (VersionRecord entry : old) {
if (!entry.id().equals("null") || entry.marker()) continue;
discarded = entry;
try (DataInputStream input = new DataInputStream(Files.newInputStream(
versionPath(bucket, key, entry.storageId())))) {
replaced = readRecord(input).metadata().length();
}
}
}
if (maxTotal - (used - replaced) < metadata.length())
throw new StoreException(507, "InsufficientStorage", "Store capacity limit reached");
Path target = versionPath(bucket, key, storageId);
Files.createDirectories(target.getParent());
syncDirectory(target.getParent().getParent());
Files.move(pending, target, StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING);
syncDirectory(target.getParent());
List<VersionRecord> next = new ArrayList<>();
next.add(new VersionRecord(id, storageId, false, metadata.modified()));
for (VersionRecord entry : old) if (!entry.id().equals(id)) next.add(entry);
saveHistory(bucket, key, next);
used += metadata.length();
objectCount++;
Metadata current = withVersion(metadata, id, bucket, key);
index.put(indexKey(bucket, key), current);
if (discarded != null) removeStoredVersion(bucket, key, discarded, replaced);
return current;
}
private void removeStoredVersion(String bucket, String key, VersionRecord entry, long length) throws IOException {
Path file = versionPath(bucket, key, entry.storageId());
if (entry.storageId().equals("legacy")) {
try (DataInputStream input = new DataInputStream(Files.newInputStream(file))) {
if (readRecord(input).metadata().key() == null) legacyCount--;
}
}
Files.deleteIfExists(file);
syncDirectory(file.getParent());
used -= length;
objectCount--;
}
public OpenObject open(String bucket, String key) throws IOException {
return open(bucket, key, null);
}
@Override public OpenObject open(String bucket, String key, String versionId) throws IOException {
Path destination = object(bucket, key);
synchronized (lock(destination)) {
VersionRecord selected;
synchronized (this) {
List<VersionRecord> entries = history(bucket, key);
selected = versionId == null ? (entries.isEmpty() ? null : entries.getFirst()) :
entries.stream().filter(entry -> entry.id().equals(versionId)).findFirst().orElse(null);
}
if (selected == null)
throw new StoreException(404, versionId == null ? "NoSuchKey" : "NoSuchVersion",
"Object version not found");
if (selected.marker())
throw StoreException.deletedVersion(selected.id(), selected.modified(), versionId != null);
final DataInputStream input;
try { input = new DataInputStream(Files.newInputStream(destination)); }
try { input = new DataInputStream(Files.newInputStream(
versionPath(bucket, key, selected.storageId()))); }
catch (NoSuchFileException e) { throw new StoreException(404, "NoSuchKey", "Object not found"); }
try { return new OpenObject(readRecord(input).metadata(), input); }
try {
Bucket configured = buckets.get(bucket);
String exposedId = histories.containsKey(indexKey(bucket, key)) ||
configured != null && configured.versioning() != VersioningState.NEVER ? selected.id() : null;
return new OpenObject(withVersion(readRecord(input).metadata(), exposedId, bucket, key), input);
}
catch (IOException e) {
input.close();
throw e;
@@ -214,9 +609,19 @@ final class DiskStore implements ObjectStorage {
}
public void delete(String bucket, String key) throws IOException {
delete(bucket, key, null);
}
@Override public DeleteResult delete(String bucket, String key, String versionId) throws IOException {
Path destination = object(bucket, key);
synchronized (lock(destination)) {
if (!Files.exists(destination)) return;
synchronized (this) {
Bucket configured = buckets.get(bucket);
VersioningState state = configured == null ? VersioningState.NEVER : configured.versioning();
if (state != VersioningState.NEVER || versionId != null)
return deleteVersioned(bucket, key, versionId, state);
}
if (!Files.exists(destination)) return new DeleteResult(null, false);
long length;
boolean legacy;
try (var input = new DataInputStream(Files.newInputStream(destination))) {
@@ -232,9 +637,146 @@ final class DiskStore implements ObjectStorage {
index.remove(indexKey(bucket, key));
syncDirectory(destination.getParent());
}
return new DeleteResult(null, false);
}
}
private DeleteResult deleteVersioned(String bucket, String key, String versionId,
VersioningState state) throws IOException {
List<VersionRecord> old = history(bucket, key);
List<VersionRecord> next = new ArrayList<>();
VersionRecord removed = null;
if (versionId != null) {
for (VersionRecord entry : old) {
if (entry.id().equals(versionId)) removed = entry;
else next.add(entry);
}
if (removed == null)
throw new StoreException(404, "NoSuchVersion", "Object version not found");
} else {
String id = state == VersioningState.ENABLED ? UUID.randomUUID().toString() : "null";
next.add(new VersionRecord(id, "", true, Instant.now().toEpochMilli()));
for (VersionRecord entry : old) {
if (state == VersioningState.SUSPENDED && entry.id().equals("null")) removed = entry;
else next.add(entry);
}
}
saveHistory(bucket, key, next);
updateCurrentIndex(bucket, key, next);
if (removed != null && !removed.marker()) {
Path file = versionPath(bucket, key, removed.storageId());
long length;
try (DataInputStream input = new DataInputStream(Files.newInputStream(file))) {
length = readRecord(input).metadata().length();
}
removeStoredVersion(bucket, key, removed, length);
}
if (versionId != null) return new DeleteResult(versionId, removed.marker());
return new DeleteResult(next.getFirst().id(), true);
}
private void updateCurrentIndex(String bucket, String key, List<VersionRecord> entries) throws IOException {
if (entries.isEmpty() || entries.getFirst().marker()) {
index.remove(indexKey(bucket, key));
return;
}
VersionRecord latest = entries.getFirst();
try (DataInputStream input = new DataInputStream(Files.newInputStream(
versionPath(bucket, key, latest.storageId())))) {
index.put(indexKey(bucket, key),
withVersion(readRecord(input).metadata(), latest.id(), bucket, key));
}
}
@Override public Map<String, String> tags(String bucket, String key) throws IOException {
return tags(bucket, key, null);
}
@Override public Map<String, String> tags(String bucket, String key, String versionId) throws IOException {
try (OpenObject object = open(bucket, key, versionId)) { return object.metadata().tags(); }
}
@Override public void setTags(String bucket, String key, Map<String, String> tags) throws IOException {
setTags(bucket, key, null, tags);
}
@Override public void setTags(String bucket, String key, String versionId,
Map<String, String> tags) throws IOException {
rewriteAttributes(bucket, key, versionId, tags, null);
}
@Override public void setObjectAcl(String bucket, String key, String versionId,
Map<String, String> acl) throws IOException {
rewriteAttributes(bucket, key, versionId, null, acl);
}
private void rewriteAttributes(String bucket, String key, String versionId,
Map<String, String> tags, Map<String, String> acl) throws IOException {
Path destination = object(bucket, key);
synchronized (lock(destination)) {
VersionRecord selected;
boolean current;
synchronized (this) {
List<VersionRecord> entries = history(bucket, key);
selected = versionId == null ? (entries.isEmpty() ? null : entries.getFirst()) :
entries.stream().filter(entry -> entry.id().equals(versionId)).findFirst().orElse(null);
current = selected != null && !entries.isEmpty() && selected == entries.getFirst();
}
if (selected == null || selected.marker())
throw new StoreException(404, versionId == null ? "NoSuchKey" : "NoSuchVersion",
"Object version not found");
destination = versionPath(bucket, key, selected.storageId());
Path pending = Files.createTempFile(temporary, "tags-", ".part");
try {
try (DataInputStream input = new DataInputStream(Files.newInputStream(destination));
OutputStream output = Files.newOutputStream(pending)) {
Metadata old = readRecord(input).metadata();
if (old.key() == null) throw new StoreException(501, "NotImplemented", "Legacy object tags are unsupported");
Metadata updated = new Metadata(old.length(), old.modified(), old.etag(), old.sha256(),
bucket, key, old.contentType(), old.userMetadata(),
tags == null ? old.tags() : Map.copyOf(tags),
old.versionId(), old.checksums(),
acl == null ? old.acl() : Map.copyOf(acl));
output.write(recordHeader(updated));
if (input.transferTo(output) != old.length()) throw new IOException("Object length changed during tag update");
}
try (FileChannel channel = FileChannel.open(pending, StandardOpenOption.WRITE)) { channel.force(true); }
synchronized (this) {
Files.move(pending, destination, StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING);
if (current) {
Metadata old = index.get(indexKey(bucket, key));
index.put(indexKey(bucket, key), new Metadata(old.length(), old.modified(), old.etag(),
old.sha256(), bucket, key, old.contentType(), old.userMetadata(),
tags == null ? old.tags() : Map.copyOf(tags),
old.versionId(), old.checksums(),
acl == null ? old.acl() : Map.copyOf(acl)));
}
syncDirectory(destination.getParent());
}
} finally { Files.deleteIfExists(pending); }
}
}
private static byte[] recordHeader(Metadata metadata) {
byte[] bucket = metadata.bucket().getBytes(StandardCharsets.UTF_8);
byte[] key = metadata.key().getBytes(StandardCharsets.UTF_8);
byte[] type = metadata.contentType().getBytes(StandardCharsets.UTF_8);
byte[] custom = ObjectAttributes.encode(metadata.userMetadata(), 4096);
byte[] tags = ObjectAttributes.encode(metadata.tags(), 8192);
byte[] checksums = ObjectAttributes.encode(metadata.checksums(), CHECKSUM_AREA);
byte[] acl = ObjectAttributes.encode(metadata.acl(), ACL_AREA);
ByteBuffer header = ByteBuffer.allocate(HEADER_V5 + acl.length +
bucket.length + key.length + type.length + custom.length + tags.length)
.putLong(MAGIC_V5).putLong(metadata.length()).putLong(metadata.modified())
.put(HexFormat.of().parseHex(metadata.etag())).put(metadata.sha256())
.putShort((short) bucket.length).putShort((short) key.length).putShort((short) type.length)
.putShort((short) custom.length).putShort((short) tags.length)
.putShort((short) checksums.length);
header.position(header.position() + CHECKSUM_AREA - checksums.length);
header.put(checksums).putShort((short) acl.length);
return header.put(acl).put(bucket).put(key).put(type).put(custom).put(tags).array();
}
public synchronized ListPage list(String bucket, String prefix, String delimiter, int maxKeys, String after) {
List<ListedObject> entries = new ArrayList<>();
List<String> prefixes = new ArrayList<>();
@@ -271,9 +813,58 @@ final class DiskStore implements ObjectStorage {
return new ListPage(entries, prefixes, truncated ? lastKey : null, truncated);
}
@Override public synchronized VersionPage listVersions(String bucket, String prefix,
String keyMarker, String versionMarker,
int maxKeys) throws IOException {
bucket(bucket);
if (versionMarker != null && keyMarker == null)
throw new StoreException(400, "InvalidArgument", "Version marker requires a key marker");
if (maxKeys == 0) return new VersionPage(List.of(), null, null, false);
NavigableSet<String> keys = new TreeSet<>();
for (String name : histories.keySet()) if (name.startsWith(bucket + "\0")) keys.add(name.substring(bucket.length() + 1));
for (Metadata meta : index.values()) if (bucket.equals(meta.bucket())) keys.add(meta.key());
List<VersionEntry> page = new ArrayList<>();
String nextKey = null;
String nextVersion = null;
boolean truncated = false;
for (String key : keys) {
if (!key.startsWith(prefix) || keyMarker != null && key.compareTo(keyMarker) < 0) continue;
List<VersionRecord> entries = history(bucket, key);
boolean pastMarker = keyMarker == null || !key.equals(keyMarker) || versionMarker == null;
for (int i = 0; i < entries.size(); i++) {
VersionRecord entry = entries.get(i);
if (keyMarker != null && key.equals(keyMarker)) {
if (versionMarker == null) continue;
if (!pastMarker) {
if (entry.id().equals(versionMarker)) pastMarker = true;
continue;
}
}
if (page.size() == maxKeys) {
truncated = true;
break;
}
Metadata meta = null;
if (!entry.marker()) {
try (DataInputStream input = new DataInputStream(Files.newInputStream(
versionPath(bucket, key, entry.storageId())))) {
meta = withVersion(readRecord(input).metadata(), entry.id(), bucket, key);
}
}
page.add(new VersionEntry(key, entry.id(), entry.modified(), entry.marker(), i == 0, meta));
nextKey = key;
nextVersion = entry.id();
}
if (truncated) break;
}
return new VersionPage(page, truncated ? nextKey : null, truncated ? nextVersion : null, truncated);
}
static Record readRecord(DataInputStream in) throws IOException {
long magic = in.readLong();
if (magic != MAGIC_V1 && magic != MAGIC_V2) throw new IOException("Invalid object record");
if (magic != MAGIC_V1 && magic != MAGIC_V2 && magic != MAGIC_V3 && magic != MAGIC_V4 &&
magic != MAGIC_V5)
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);
@@ -283,8 +874,30 @@ final class DiskStore implements ObjectStorage {
return new Record(new Metadata(length, modified, SigV4.hex(md5), sha,
null, null, "application/octet-stream"), HEADER_V1);
int bucketLength = in.readUnsignedShort(), keyLength = in.readUnsignedShort(), typeLength = in.readUnsignedShort();
int metadataLength = magic == MAGIC_V3 || magic == MAGIC_V4 || magic == MAGIC_V5 ? in.readUnsignedShort() : 0;
int tagsLength = magic == MAGIC_V3 || magic == MAGIC_V4 || magic == MAGIC_V5 ? in.readUnsignedShort() : 0;
int checksumLength = magic == MAGIC_V4 || magic == MAGIC_V5 ? in.readUnsignedShort() : 0;
if (bucketLength < 1 || bucketLength > 63 || keyLength < 1 || keyLength > 1024 || typeLength < 1 || typeLength > 255)
throw new IOException("Invalid object record metadata");
if (metadataLength > 4096 || tagsLength > 8192 || checksumLength > CHECKSUM_AREA)
throw new IOException("Invalid object attributes");
Map<String, String> checksums = Map.of();
if (magic == MAGIC_V4 || magic == MAGIC_V5) {
byte[] area = in.readNBytes(CHECKSUM_AREA);
if (area.length != CHECKSUM_AREA) throw new IOException("Truncated checksum attributes");
checksums = ObjectAttributes.decode(Arrays.copyOfRange(area,
CHECKSUM_AREA - checksumLength, CHECKSUM_AREA));
}
Map<String, String> acl = Map.of();
int aclLength = 0;
if (magic == MAGIC_V5) {
int size = in.readUnsignedShort();
if (size > ACL_AREA) throw new IOException("Invalid object ACL");
byte[] bytes = in.readNBytes(size);
if (bytes.length != size) throw new IOException("Truncated object ACL");
acl = ObjectAttributes.decode(bytes);
aclLength = size;
}
String bucket = utf8(in.readNBytes(bucketLength));
String key = utf8(in.readNBytes(keyLength));
String contentType = utf8(in.readNBytes(typeLength));
@@ -292,8 +905,13 @@ final class DiskStore implements ObjectStorage {
key.getBytes(StandardCharsets.UTF_8).length != keyLength ||
contentType.getBytes(StandardCharsets.UTF_8).length != typeLength)
throw new IOException("Invalid object record metadata");
Map<String, String> metadata = ObjectAttributes.decode(in.readNBytes(metadataLength));
Map<String, String> tags = ObjectAttributes.decode(in.readNBytes(tagsLength));
return new Record(new Metadata(length, modified, SigV4.hex(md5), sha,
bucket, key, contentType), HEADER_V2 + bucketLength + keyLength + typeLength);
bucket, key, contentType, metadata, tags, null, checksums, acl),
(magic == MAGIC_V5 ? HEADER_V5 + aclLength :
magic == MAGIC_V4 ? HEADER_V4 : magic == MAGIC_V3 ? HEADER_V3 : HEADER_V2) +
bucketLength + keyLength + typeLength + metadataLength + tagsLength);
}
private static String utf8(byte[] bytes) throws IOException {
return StandardCharsets.UTF_8.newDecoder().onMalformedInput(CodingErrorAction.REPORT)
File diff suppressed because it is too large. Load diff
+11 -1
View File
@@ -3,6 +3,7 @@ package cloud.lunarsky.store;
import java.io.IOException;
import java.io.InputStream;
import java.util.List;
import java.util.Map;
interface MultipartStorage {
record Part(int number, String etag) {}
@@ -10,7 +11,16 @@ interface MultipartStorage {
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 create(String bucket, String key, String contentType,
Map<String, String> userMetadata, Map<String, String> tags,
Map<String, String> acl) throws IOException;
default String create(String bucket, String key, String contentType,
Map<String, String> userMetadata, Map<String, String> tags) throws IOException {
return create(bucket, key, contentType, userMetadata, tags, Map.of());
}
default String create(String bucket, String key, String contentType) throws IOException {
return create(bucket, key, contentType, Map.of(), Map.of());
}
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;
+30 -7
View File
@@ -8,12 +8,16 @@ import cloud.lunarsky.store.MultipartStorage.Part;
final class MultipartStore implements MultipartStorage {
private static final int MAGIC = 0x4c534d50;
private static final int MAGIC_V2 = 0x4c534d51;
private static final int MAGIC_V3 = 0x4c534d52;
private final DiskStore store;
private final Path root;
private long staged;
private int active;
private record Upload(String bucket, String key, String contentType) {}
private record Upload(String bucket, String key, String contentType,
Map<String, String> userMetadata, Map<String, String> tags,
Map<String, String> acl) {}
MultipartStore(DiskStore store) throws IOException {
this.store = store;
@@ -42,17 +46,29 @@ final class MultipartStore implements MultipartStorage {
if (staged > store.maxTotal()) throw new IOException("Multipart staging limit exceeded");
}
public synchronized String create(String bucket, String key, String contentType) throws IOException {
public synchronized String create(String bucket, String key, String contentType,
Map<String, String> userMetadata, Map<String, String> tags,
Map<String, String> acl) throws IOException {
if (Files.exists(store.root().resolve("buckets.bin"))) store.bucket(bucket);
if (active >= 32) throw new StoreException(503, "SlowDown", "Too many active uploads");
String id = UUID.randomUUID().toString();
Path pending = root.resolve(".creating-" + id), dir = root.resolve(id);
Files.createDirectory(pending);
try {
try (var output = new DataOutputStream(Files.newOutputStream(pending.resolve("manifest"), StandardOpenOption.CREATE_NEW))) {
output.writeInt(MAGIC);
output.writeInt(MAGIC_V3);
output.writeUTF(bucket);
output.writeUTF(key);
output.writeUTF(contentType);
byte[] custom = ObjectAttributes.encode(userMetadata, 4096);
byte[] encodedTags = ObjectAttributes.encode(tags, 8192);
output.writeShort(custom.length);
output.write(custom);
output.writeShort(encodedTags.length);
output.write(encodedTags);
byte[] encodedAcl = ObjectAttributes.encode(acl, 2048);
output.writeShort(encodedAcl.length);
output.write(encodedAcl);
}
try (var channel = java.nio.channels.FileChannel.open(pending.resolve("manifest"), StandardOpenOption.READ)) {
channel.force(true);
@@ -109,7 +125,7 @@ final class MultipartStore implements MultipartStorage {
}
if (count != length) throw new StoreException(400, "IncompleteBody", "Part length does not match Content-Length");
byte[] actual = sha.digest();
if (!MessageDigest.isEqual(actual, HexFormat.of().parseHex(expectedHash)))
if (expectedHash != null && !MessageDigest.isEqual(actual, HexFormat.of().parseHex(expectedHash)))
throw new StoreException(400, "XAmzContentSHA256Mismatch", "Part hash mismatch");
if (checksum != null && !Base64.getEncoder().encodeToString(actual).equals(checksum))
throw new StoreException(400, "BadDigest", "SHA-256 checksum mismatch");
@@ -153,8 +169,9 @@ final class MultipartStore implements MultipartStorage {
}
ObjectStorage.Metadata result;
try (InputStream input = new PartsInput(paths)) {
Upload upload = readUpload(dir);
result = store.put(bucket, key, input, total, SigV4.hex(sha.digest()), null, false,
readUpload(dir).contentType());
upload.contentType(), upload.userMetadata(), upload.tags(), Map::of, upload.acl());
}
remove(dir);
return result;
@@ -217,8 +234,14 @@ final class MultipartStore implements MultipartStorage {
}
private static Upload readUpload(Path dir) throws IOException {
try (var input = new DataInputStream(Files.newInputStream(dir.resolve("manifest")))) {
if (input.readInt() != MAGIC) throw new IOException("Invalid multipart upload manifest");
Upload upload = new Upload(input.readUTF(), input.readUTF(), input.readUTF());
int magic = input.readInt();
if (magic != MAGIC && magic != MAGIC_V2 && magic != MAGIC_V3)
throw new IOException("Invalid multipart upload manifest");
String bucket = input.readUTF(), key = input.readUTF(), type = input.readUTF();
Map<String, String> custom = magic != MAGIC ? ObjectAttributes.decode(input.readNBytes(input.readUnsignedShort())) : Map.of();
Map<String, String> tags = magic != MAGIC ? ObjectAttributes.decode(input.readNBytes(input.readUnsignedShort())) : Map.of();
Map<String, String> acl = magic == MAGIC_V3 ? ObjectAttributes.decode(input.readNBytes(input.readUnsignedShort())) : Map.of();
Upload upload = new Upload(bucket, key, type, custom, tags, acl);
if (input.read() != -1) throw new IOException("Invalid multipart upload manifest");
return upload;
}
+57 -11
View File
@@ -14,6 +14,8 @@ import java.util.HexFormat;
import java.util.List;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
final class NodeClient {
record Node(UUID id, UUID hostId, URI url) {}
@@ -25,6 +27,10 @@ final class NodeClient {
private final String token;
private final String repairToken;
private final HttpClient http = HttpClient.newBuilder().connectTimeout(Duration.ofSeconds(3)).build();
private final ConcurrentHashMap<UUID, Long> unreadableUntil = new ConcurrentHashMap<>();
private final ConcurrentHashMap<UUID, Long> healthyUntil = new ConcurrentHashMap<>();
private static final long READ_RETRY_NANOS = TimeUnit.SECONDS.toNanos(5);
private static final long HEALTH_FRESH_NANOS = TimeUnit.SECONDS.toNanos(3);
NodeClient(List<Node> nodes, String token, String repairToken) {
if (nodes.isEmpty() || nodes.stream().map(Node::id).distinct().count() != nodes.size() ||
@@ -97,14 +103,33 @@ final class NodeClient {
for (int i = 0; i < nodes.size(); i++) {
Node node = nodes.get(i);
NodeIdentity actual = probeIfAvailable(node.url(), token);
if (actual == null) continue;
if (actual.nodeId().equals(node.id()) && actual.hostId().equals(node.hostId()))
healthy.add(faultDomain(i, testNodeDomains));
if (actual == null || !actual.nodeId().equals(node.id()) || !actual.hostId().equals(node.hostId())) {
markUnreadable(node);
continue;
}
unreadableUntil.remove(node.id());
healthyUntil.put(node.id(), System.nanoTime() + HEALTH_FRESH_NANOS);
healthy.add(faultDomain(i, testNodeDomains));
if (healthy.size() >= required) return true;
}
return false;
}
private void markUnreadable(Node node) {
healthyUntil.remove(node.id());
unreadableUntil.put(node.id(), System.nanoTime() + READ_RETRY_NANOS);
}
private boolean unreadable(Node node) {
Long until = unreadableUntil.get(node.id());
return until != null && System.nanoTime() - until < 0;
}
private boolean recentlyHealthy(Node node) {
Long until = healthyUntil.get(node.id());
return until != null && System.nanoTime() - until < 0;
}
void put(int index, UUID id, byte[] data, byte[] sha256) throws IOException {
put(index, id, data, sha256, false);
}
@@ -132,18 +157,39 @@ final class NodeClient {
byte[] get(int index, UUID id, int length, byte[] sha256) throws IOException {
if (length < 1 || length > ClusterNode.MAX_SEGMENT) throw new IOException("Invalid segment length");
Node node = nodes.get(index);
if (unreadable(node)) throw new IOException("Storage node is temporarily unreachable");
if (!recentlyHealthy(node)) {
NodeIdentity actual = probeIfAvailable(node.url(), token);
if (actual == null || !actual.nodeId().equals(node.id()) || !actual.hostId().equals(node.hostId())) {
markUnreadable(node);
throw new IOException("Storage node is temporarily unreachable");
}
healthyUntil.put(node.id(), System.nanoTime() + HEALTH_FRESH_NANOS);
}
HttpRequest request = HttpRequest.newBuilder(node.url().resolve("/segments/" + id))
.timeout(Duration.ofSeconds(30)).header("X-Cluster-Token", token)
.header("X-Cluster-Expected-Node", node.id().toString()).GET().build();
HttpResponse<InputStream> response = send(request, HttpResponse.BodyHandlers.ofInputStream());
try (InputStream body = response.body()) {
if (response.statusCode() != 200)
throw new IOException("Node " + node.id() + " has no verified copy of segment " + id);
byte[] bytes = body.readNBytes(length + 1);
if (bytes.length != length || !MessageDigest.isEqual(SigV4.hash(bytes), sha256))
throw new IOException("Node " + node.id() + " has no verified copy of segment " + id);
return bytes;
HttpResponse<InputStream> response;
try { response = send(request, HttpResponse.BodyHandlers.ofInputStream()); }
catch (IOException error) {
markUnreadable(node);
throw error;
}
if (response.statusCode() != 200) {
response.body().close();
throw new IOException("Node " + node.id() + " has no verified copy of segment " + id);
}
byte[] bytes;
try (InputStream body = response.body()) { bytes = body.readNBytes(length + 1); }
catch (IOException error) {
markUnreadable(node);
throw error;
}
if (bytes.length != length || !MessageDigest.isEqual(SigV4.hash(bytes), sha256))
throw new IOException("Node " + node.id() + " has no verified copy of segment " + id);
unreadableUntil.remove(node.id());
healthyUntil.put(node.id(), System.nanoTime() + HEALTH_FRESH_NANOS);
return bytes;
}
List<StoredSegment> inventory(int index, String shard, UUID after) throws IOException {
@@ -0,0 +1,114 @@
package cloud.lunarsky.store;
import com.sun.net.httpserver.Headers;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.TreeMap;
final class ObjectAttributes {
private ObjectAttributes() {}
static Map<String, String> userMetadata(Headers headers) {
Map<String, String> values = new TreeMap<>();
int total = 0;
for (String name : headers.keySet()) {
if (!name.toLowerCase(java.util.Locale.ROOT).startsWith("x-amz-meta-")) continue;
String key = name.substring(11).toLowerCase(java.util.Locale.ROOT);
if (!key.matches("[a-z0-9][a-z0-9._-]{0,127}"))
throw new StoreException(400, "InvalidArgument", "Invalid user metadata key");
String value = SigV4.single(headers, name);
if (value == null || !value.chars().allMatch(c -> c >= 32 && c <= 126))
throw new StoreException(400, "InvalidArgument", "Invalid user metadata value");
total += key.getBytes(StandardCharsets.UTF_8).length + value.getBytes(StandardCharsets.UTF_8).length;
values.put(key, value);
}
if (total > 2048) throw new StoreException(400, "MetadataTooLarge", "User metadata exceeds 2 KiB");
return Map.copyOf(values);
}
static Map<String, String> tagsHeader(String raw) {
if (raw == null) return Map.of();
Map<String, String> tags = new TreeMap<>();
if (raw.isEmpty()) return tags;
for (String pair : raw.split("&", -1)) {
String[] parts = pair.split("=", 2);
if (parts.length != 2) throw new StoreException(400, "InvalidTag", "Invalid tagging header");
String key = SigV4.decode(parts[0]);
String value = SigV4.decode(parts[1]);
if (tags.put(key, value) != null) throw new StoreException(400, "InvalidTag", "Duplicate tag key");
}
validateTags(tags);
return Map.copyOf(tags);
}
static void validateTags(Map<String, String> tags) {
if (tags.size() > 10) throw new StoreException(400, "InvalidTag", "Too many object tags");
for (var entry : tags.entrySet()) {
if (entry.getKey().isEmpty() || entry.getKey().length() > 128 ||
entry.getValue().length() > 256 ||
!xmlText(entry.getKey()) || !xmlText(entry.getValue()))
throw new StoreException(400, "InvalidTag", "Invalid tag key or value");
}
}
private static boolean xmlText(String value) {
for (int i = 0; i < value.length();) {
int point = value.codePointAt(i);
if (point < 32 || point > 0x10ffff || point >= 0xd800 && point <= 0xdfff ||
point >= 0xfffe && point <= 0xffff) return false;
i += Character.charCount(point);
}
return true;
}
static byte[] encode(Map<String, String> values, int limit) {
try {
ByteArrayOutputStream bytes = new ByteArrayOutputStream();
DataOutputStream output = new DataOutputStream(bytes);
output.writeShort(values.size());
for (var entry : new TreeMap<>(values).entrySet()) {
byte[] key = entry.getKey().getBytes(StandardCharsets.UTF_8);
byte[] value = entry.getValue().getBytes(StandardCharsets.UTF_8);
output.writeShort(key.length);
output.writeShort(value.length);
output.write(key);
output.write(value);
}
if (bytes.size() > limit) throw new StoreException(400, "InvalidArgument", "Object attributes are too large");
return bytes.toByteArray();
} catch (IOException error) { throw new IllegalStateException(error); }
}
static Map<String, String> decode(byte[] bytes) throws IOException {
if (bytes == null || bytes.length == 0) return Map.of();
try (DataInputStream input = new DataInputStream(new ByteArrayInputStream(bytes))) {
int count = input.readUnsignedShort();
if (count > 128) throw new IOException("Invalid object attributes");
Map<String, String> values = new TreeMap<>();
for (int i = 0; i < count; i++) {
int keyLength = input.readUnsignedShort(), valueLength = input.readUnsignedShort();
String key = decodeUtf8(input.readNBytes(keyLength), keyLength);
String value = decodeUtf8(input.readNBytes(valueLength), valueLength);
if (values.put(key, value) != null) throw new IOException("Duplicate object attribute");
}
if (input.available() != 0) throw new IOException("Trailing object attributes");
return Map.copyOf(values);
}
}
private static String decodeUtf8(byte[] bytes, int length) throws IOException {
if (bytes.length != length) throw new IOException("Truncated object attributes");
try {
return StandardCharsets.UTF_8.newDecoder()
.onMalformedInput(java.nio.charset.CodingErrorAction.REPORT)
.decode(java.nio.ByteBuffer.wrap(bytes)).toString();
} catch (java.nio.charset.CharacterCodingException error) {
throw new IOException("Invalid object attributes", error);
}
}
}
+97 -2
View File
@@ -3,11 +3,46 @@ package cloud.lunarsky.store;
import java.io.IOException;
import java.io.InputStream;
import java.util.List;
import java.util.Map;
/** Storage operations shared by the local and cluster gateways. */
interface ObjectStorage extends AutoCloseable {
enum VersioningState { NEVER, ENABLED, SUSPENDED }
record Limits(long maxObjectBytes, long maxTotalBytes) {}
record Bucket(String name, long created, VersioningState versioning, Map<String, String> acl) {
Bucket(String name, long created) { this(name, created, VersioningState.NEVER, Map.of()); }
Bucket(String name, long created, VersioningState versioning) {
this(name, created, versioning, Map.of());
}
}
record Metadata(long length, long modified, String etag, byte[] sha256,
String bucket, String key, String contentType) {}
String bucket, String key, String contentType,
Map<String, String> userMetadata, Map<String, String> tags,
String versionId, Map<String, String> checksums, Map<String, String> acl) {
Metadata(long length, long modified, String etag, byte[] sha256,
String bucket, String key, String contentType,
Map<String, String> userMetadata, Map<String, String> tags,
String versionId, Map<String, String> checksums) {
this(length, modified, etag, sha256, bucket, key, contentType,
userMetadata, tags, versionId, checksums, Map.of());
}
Metadata(long length, long modified, String etag, byte[] sha256,
String bucket, String key, String contentType,
Map<String, String> userMetadata, Map<String, String> tags,
String versionId) {
this(length, modified, etag, sha256, bucket, key, contentType,
userMetadata, tags, versionId, Map.of());
}
Metadata(long length, long modified, String etag, byte[] sha256,
String bucket, String key, String contentType,
Map<String, String> userMetadata, Map<String, String> tags) {
this(length, modified, etag, sha256, bucket, key, contentType, userMetadata, tags, null);
}
Metadata(long length, long modified, String etag, byte[] sha256,
String bucket, String key, String contentType) {
this(length, modified, etag, sha256, bucket, key, contentType, Map.of(), Map.of(), null);
}
}
record OpenObject(Metadata metadata, InputStream stream) implements AutoCloseable {
public void close() throws IOException { stream.close(); }
}
@@ -15,12 +50,72 @@ interface ObjectStorage extends AutoCloseable {
record ListPage(List<ListedObject> objects, List<String> prefixes, String nextKey, boolean truncated) {
int keyCount() { return objects.size() + prefixes.size(); }
}
record VersionEntry(String key, String versionId, long modified, boolean deleteMarker,
boolean latest, Metadata metadata) {}
record VersionPage(List<VersionEntry> entries, String nextKey, String nextVersionId,
boolean truncated) {}
record DeleteResult(String versionId, boolean deleteMarker) {}
Metadata put(String bucket, String key, InputStream input, long length, String expectedHash,
String checksum, boolean createOnly, String contentType) throws IOException;
String checksum, boolean createOnly, String contentType,
Map<String, String> userMetadata, Map<String, String> tags,
java.util.function.Supplier<Map<String, String>> checksums,
Map<String, String> acl) throws IOException;
default Metadata put(String bucket, String key, InputStream input, long length, String expectedHash,
String checksum, boolean createOnly, String contentType,
Map<String, String> userMetadata, Map<String, String> tags,
java.util.function.Supplier<Map<String, String>> checksums) throws IOException {
return put(bucket, key, input, length, expectedHash, checksum, createOnly, contentType,
userMetadata, tags, checksums, Map.of());
}
default Metadata put(String bucket, String key, InputStream input, long length, String expectedHash,
String checksum, boolean createOnly, String contentType,
Map<String, String> userMetadata, Map<String, String> tags) throws IOException {
return put(bucket, key, input, length, expectedHash, checksum, createOnly, contentType,
userMetadata, tags, Map::of);
}
default Metadata put(String bucket, String key, InputStream input, long length, String expectedHash,
String checksum, boolean createOnly, String contentType) throws IOException {
return put(bucket, key, input, length, expectedHash, checksum, createOnly, contentType, Map.of(), Map.of());
}
OpenObject open(String bucket, String key) throws IOException;
default OpenObject open(String bucket, String key, String versionId) throws IOException {
if (versionId == null) return open(bucket, key);
throw new StoreException(501, "NotImplemented", "Object versioning is unavailable");
}
Map<String, String> tags(String bucket, String key) throws IOException;
default Map<String, String> tags(String bucket, String key, String versionId) throws IOException {
if (versionId == null) return tags(bucket, key);
throw new StoreException(501, "NotImplemented", "Versioned tagging is unavailable");
}
void setTags(String bucket, String key, Map<String, String> tags) throws IOException;
default void setTags(String bucket, String key, String versionId,
Map<String, String> tags) throws IOException {
if (versionId == null) setTags(bucket, key, tags);
else throw new StoreException(501, "NotImplemented", "Versioned tagging is unavailable");
}
void delete(String bucket, String key) throws IOException;
default DeleteResult delete(String bucket, String key, String versionId) throws IOException {
if (versionId != null) throw new StoreException(501, "NotImplemented", "Object versioning is unavailable");
delete(bucket, key);
return new DeleteResult(null, false);
}
default void setVersioning(String bucket, VersioningState state) throws IOException {
throw new StoreException(501, "NotImplemented", "Object versioning is unavailable");
}
default VersionPage listVersions(String bucket, String prefix, String keyMarker,
String versionMarker, int maxKeys) throws IOException {
throw new StoreException(501, "NotImplemented", "Object versioning is unavailable");
}
ListPage list(String bucket, String prefix, String delimiter, int maxKeys, String after) throws IOException;
void ensureBucket(String bucket) throws IOException;
Bucket bucket(String bucket) throws IOException;
List<Bucket> buckets() throws IOException;
void createBucket(String bucket) throws IOException;
void deleteBucket(String bucket) throws IOException;
void setBucketAcl(String bucket, Map<String, String> acl) throws IOException;
void setObjectAcl(String bucket, String key, String versionId, Map<String, String> acl) throws IOException;
Limits limits();
default boolean ready() { return true; }
void close() throws IOException;
}
+58 -1
View File
@@ -21,7 +21,7 @@ final class SchemaMigrator {
result.next();
version = result.getInt(1);
}
if (version > 4) throw new IOException("Metadata schema is newer than this ObjectStore build");
if (version > 10) 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))");
@@ -47,12 +47,69 @@ final class SchemaMigrator {
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)");
}
if (version < 5) {
statement.execute("ALTER TABLE cluster_objects ADD COLUMN user_metadata bytea");
statement.execute("ALTER TABLE cluster_objects ADD COLUMN tags bytea");
statement.execute("ALTER TABLE cluster_uploads ADD COLUMN user_metadata bytea");
statement.execute("ALTER TABLE cluster_uploads ADD COLUMN tags bytea");
statement.execute("INSERT INTO cluster_schema_migrations VALUES (5)");
}
if (version < 6) {
statement.execute("CREATE TABLE cluster_buckets (name text PRIMARY KEY, created_at bigint NOT NULL)");
statement.execute("INSERT INTO cluster_buckets SELECT DISTINCT bucket, " +
"CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS bigint) FROM cluster_usage");
statement.execute("INSERT INTO cluster_schema_migrations VALUES (6)");
}
if (version < 7) {
statement.execute("ALTER TABLE cluster_buckets ADD COLUMN versioning_state text NOT NULL " +
"DEFAULT 'NEVER' CHECK (versioning_state IN ('NEVER', 'ENABLED', 'SUSPENDED'))");
statement.execute("INSERT INTO cluster_schema_migrations VALUES (7)");
}
if (version < 8) {
statement.execute("CREATE TABLE cluster_object_versions (sequence bigint GENERATED ALWAYS AS IDENTITY, " +
"bucket text NOT NULL, object_key text COLLATE \"C\" NOT NULL, version_id text NOT NULL, " +
"delete_marker boolean NOT NULL, generation uuid, length bigint, modified bigint NOT NULL, " +
"etag text, sha256 bytea, content_type text, user_metadata bytea, tags bytea, " +
"PRIMARY KEY (bucket, object_key, version_id), " +
"CHECK (delete_marker = (generation IS NULL)))");
statement.execute("CREATE INDEX cluster_versions_order ON cluster_object_versions " +
"(bucket, object_key, sequence DESC)");
statement.execute("CREATE TABLE cluster_object_heads (bucket text NOT NULL, " +
"object_key text COLLATE \"C\" NOT NULL, version_id text NOT NULL, " +
"PRIMARY KEY (bucket, object_key), " +
"FOREIGN KEY (bucket, object_key, version_id) REFERENCES cluster_object_versions " +
"(bucket, object_key, version_id) DEFERRABLE INITIALLY DEFERRED)");
statement.execute("INSERT INTO cluster_object_versions " +
"(bucket, object_key, version_id, delete_marker, generation, length, modified, etag, " +
"sha256, content_type, user_metadata, tags) " +
"SELECT bucket, object_key, 'null', false, generation, length, modified, etag, sha256, " +
"content_type, user_metadata, tags FROM cluster_objects");
statement.execute("INSERT INTO cluster_object_heads " +
"SELECT bucket, object_key, 'null' FROM cluster_objects");
statement.execute("INSERT INTO cluster_schema_migrations VALUES (8)");
}
if (version < 9) {
statement.execute("ALTER TABLE cluster_object_versions ADD COLUMN checksum_metadata bytea");
statement.execute("INSERT INTO cluster_schema_migrations VALUES (9)");
}
if (version < 10) {
statement.execute("ALTER TABLE cluster_buckets ADD COLUMN acl bytea");
statement.execute("ALTER TABLE cluster_object_versions ADD COLUMN acl bytea");
statement.execute("ALTER TABLE cluster_uploads ADD COLUMN acl bytea");
statement.execute("INSERT INTO cluster_schema_migrations VALUES (10)");
}
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")) {
insert.setString(1, bucket);
insert.executeUpdate();
}
try (PreparedStatement insert = connection.prepareStatement(
"INSERT INTO cluster_buckets (name, created_at) VALUES (?, ?) ON CONFLICT DO NOTHING")) {
insert.setString(1, bucket);
insert.setLong(2, System.currentTimeMillis());
insert.executeUpdate();
}
int format;
try (Statement statement = connection.createStatement();
ResultSet result = statement.executeQuery("SELECT version FROM cluster_format WHERE singleton=1")) {
+104 -11
View File
@@ -12,39 +12,124 @@ import java.time.format.DateTimeFormatter;
import java.util.Arrays;
import java.util.HexFormat;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.regex.Pattern;
import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
final class SigV4 {
record Verified(String payload, String applicationQuery, byte[] signingKey,
String date, String scope, String signature, String principal) {
Verified(String payload, String applicationQuery, byte[] signingKey,
String date, String scope, String signature) {
this(payload, applicationQuery, signingKey, date, scope, signature, null);
}
boolean streaming() { return payload.startsWith("STREAMING-AWS4-HMAC-SHA256-PAYLOAD"); }
}
private static final DateTimeFormatter DATE = DateTimeFormatter.ofPattern("uuuuMMdd'T'HHmmss'Z'").withZone(ZoneOffset.UTC);
private static final Pattern HEX = Pattern.compile("[0-9a-f]{64}");
private final String accessKey, secretKey, region;
private final Map<String, String> identities;
private final String root;
private final String region;
private final Clock clock;
SigV4(String accessKey, String secretKey, String region, Clock clock) {
this.accessKey = accessKey;
this.secretKey = secretKey;
this(Map.of(accessKey, secretKey), accessKey, region, clock);
}
SigV4(Map<String, String> identities, String root, String region, Clock clock) {
this.identities = Map.copyOf(identities);
if (!this.identities.containsKey(root)) throw new IllegalArgumentException("Missing root identity");
this.root = root;
this.region = region;
this.clock = clock;
}
String root() { return root; }
Set<String> identities() { return identities.keySet(); }
String verify(String method, URI uri, Headers headers) {
return verifyRequest(method, uri, headers).payload();
}
Verified verifyRequest(String method, URI uri, Headers headers) {
if (hasPresignedQuery(uri.getRawQuery())) return verifyPresigned(method, uri, headers);
Map<String, String> fields = authorizationFields(headers);
String[] credential = credentialScope(fields.get("Credential"));
String date = signingDate(headers, credential[1]);
String payload = payloadHash(headers);
String signedHeaders = fields.get("SignedHeaders");
String canonicalHeaders = canonicalHeaders(headers, signedHeaders);
String canonicalHeaders = canonicalHeaders(headers, signedHeaders,
Set.of("host", "x-amz-date", "x-amz-content-sha256"));
String canonical = method + "\n" + encode(decode(uri.getRawPath()), true) + "\n"
+ canonicalQuery(uri.getRawQuery()) + "\n" + canonicalHeaders + "\n" + signedHeaders + "\n" + payload;
String scope = String.join("/", Arrays.copyOfRange(credential, 1, 5));
String toSign = "AWS4-HMAC-SHA256\n" + date + "\n" + scope + "\n" + hex(hash(canonical.getBytes(StandardCharsets.UTF_8)));
byte[] signingKey = signingKey(secretKey, credential[1], region);
byte[] signingKey = signingKey(secret(credential[0]), credential[1], region);
String signature = fields.get("Signature");
if (!HEX.matcher(signature).matches() || !MessageDigest.isEqual(hmac(signingKey, toSign), HexFormat.of().parseHex(signature))) denied("Signature mismatch");
return payload;
return new Verified(payload, uri.getRawQuery(), signingKey, date, scope, signature, credential[0]);
}
private Verified verifyPresigned(String method, URI uri, Headers headers) {
if (headers.containsKey("authorization")) denied("Use one authentication method");
Map<String, String> fields = new TreeMap<>();
StringBuilder application = new StringBuilder();
StringBuilder signed = new StringBuilder();
for (String part : uri.getRawQuery().split("&", -1)) {
String[] pair = part.split("=", 2);
String name = decode(pair[0]);
String value = decode(pair.length == 2 ? pair[1] : "");
if (name.startsWith("X-Amz-")) {
if (fields.put(name, value) != null) denied("Duplicate presigned parameter");
if (!name.equals("X-Amz-Signature")) appendQuery(signed, part);
} else {
appendQuery(application, part);
appendQuery(signed, part);
}
}
if (!fields.keySet().equals(Set.of("X-Amz-Algorithm", "X-Amz-Credential", "X-Amz-Date",
"X-Amz-Expires", "X-Amz-SignedHeaders", "X-Amz-Signature")) ||
!"AWS4-HMAC-SHA256".equals(fields.get("X-Amz-Algorithm")))
denied("Invalid presigned parameters");
String[] credential = credentialScope(fields.get("X-Amz-Credential"));
String date = fields.get("X-Amz-Date");
if (!date.matches("[0-9]{8}T[0-9]{6}Z") || !date.startsWith(credential[1]))
denied("Invalid signing date");
long expires;
try { expires = Long.parseLong(fields.get("X-Amz-Expires")); }
catch (NumberFormatException error) { denied("Invalid presigned expiry"); return null; }
if (expires < 1 || expires > 604800) denied("Invalid presigned expiry");
try {
Instant start = Instant.from(DATE.parse(date));
Instant now = clock.instant();
if (now.isBefore(start.minus(Duration.ofMinutes(5))) || now.isAfter(start.plusSeconds(expires)))
denied("Presigned URL has expired or is not yet valid");
} catch (java.time.DateTimeException error) { denied("Invalid signing date"); }
String signedHeaders = fields.get("X-Amz-SignedHeaders");
String canonicalHeaders = canonicalHeaders(headers, signedHeaders, Set.of("host"));
String scope = String.join("/", Arrays.copyOfRange(credential, 1, 5));
String canonical = method + "\n" + encode(decode(uri.getRawPath()), true) + "\n"
+ canonicalQuery(signed.toString()) + "\n" + canonicalHeaders + "\n"
+ signedHeaders + "\nUNSIGNED-PAYLOAD";
String toSign = "AWS4-HMAC-SHA256\n" + date + "\n" + scope + "\n"
+ hex(hash(canonical.getBytes(StandardCharsets.UTF_8)));
String signature = fields.get("X-Amz-Signature");
byte[] key = signingKey(secret(credential[0]), credential[1], region);
if (!HEX.matcher(signature).matches() ||
!MessageDigest.isEqual(hmac(key, toSign), HexFormat.of().parseHex(signature)))
denied("Signature mismatch");
return new Verified("UNSIGNED-PAYLOAD", application.toString(), key, date, scope, signature, credential[0]);
}
private static boolean hasPresignedQuery(String raw) {
return raw != null && (raw.startsWith("X-Amz-Algorithm=") || raw.contains("&X-Amz-Algorithm="));
}
private static void appendQuery(StringBuilder target, String part) {
if (!target.isEmpty()) target.append('&');
target.append(part);
}
private static Map<String, String> authorizationFields(Headers headers) {
@@ -61,11 +146,17 @@ final class SigV4 {
private String[] credentialScope(String value) {
String[] credential = value.split("/", -1);
if (credential.length != 5 || !credential[0].equals(accessKey) || !credential[2].equals(region)
if (credential.length != 5 || !identities.containsKey(credential[0]) || !credential[2].equals(region)
|| !credential[3].equals("s3") || !credential[4].equals("aws4_request")) denied("Invalid credential scope");
return credential;
}
private String secret(String accessKey) {
String secret = identities.get(accessKey);
if (secret == null) denied("Invalid credential scope");
return secret;
}
private String signingDate(Headers headers, String credentialDate) {
String date = single(headers, "x-amz-date");
if (date == null || !credentialDate.matches("[0-9]{8}") || !date.matches("[0-9]{8}T[0-9]{6}Z") || !date.startsWith(credentialDate)) denied("Invalid signing date");
@@ -79,17 +170,19 @@ final class SigV4 {
private static String payloadHash(Headers headers) {
String payload = single(headers, "x-amz-content-sha256");
if (payload == null || !HEX.matcher(payload).matches())
throw new StoreException(400, "NotImplemented", "A hexadecimal SHA-256 payload hash is required; unsigned and chunk-signed payloads are unsupported");
if (payload == null || !(HEX.matcher(payload).matches() ||
payload.equals("STREAMING-AWS4-HMAC-SHA256-PAYLOAD") ||
payload.equals("STREAMING-AWS4-HMAC-SHA256-PAYLOAD-TRAILER")))
throw new StoreException(400, "NotImplemented", "Unsupported SHA-256 payload mode");
if (headers.containsKey("x-amz-security-token")) denied("Temporary credentials are unsupported");
return payload;
}
private static String canonicalHeaders(Headers headers, String signedHeaders) {
private static String canonicalHeaders(Headers headers, String signedHeaders, Set<String> required) {
String[] names = signedHeaders.split(";", -1);
if (names.length > 32 || !signedHeaders.equals(String.join(";", Arrays.stream(names).distinct().sorted().toList()))) denied("Signed headers must be unique and sorted");
var namesSet = java.util.Set.copyOf(Arrays.asList(names));
if (!namesSet.containsAll(java.util.Set.of("host", "x-amz-date", "x-amz-content-sha256"))) denied("Missing signed headers");
if (!namesSet.containsAll(required)) denied("Missing signed headers");
for (String key : headers.keySet()) {
String lower = key.toLowerCase(java.util.Locale.ROOT);
if (lower.startsWith("x-amz-") && !namesSet.contains(lower)) denied("Unsigned Amazon header");
@@ -3,6 +3,9 @@ package cloud.lunarsky.store;
final class StoreException extends RuntimeException {
final int status;
final String code;
final String versionId;
final long modified;
final boolean deleteMarker;
StoreException(int status, String code, String message) {
this(status, code, message, null);
}
@@ -10,5 +13,20 @@ final class StoreException extends RuntimeException {
super(message, cause);
this.status = status;
this.code = code;
this.versionId = null;
this.modified = -1;
this.deleteMarker = false;
}
private StoreException(int status, String code, String message, String versionId, long modified) {
super(message);
this.status = status;
this.code = code;
this.versionId = versionId;
this.modified = modified;
this.deleteMarker = true;
}
static StoreException deletedVersion(String versionId, long modified, boolean explicit) {
return new StoreException(explicit ? 405 : 404, explicit ? "MethodNotAllowed" : "NoSuchKey",
"Object is deleted", versionId, modified);
}
}
+28 -4
View File
@@ -41,8 +41,16 @@ final class UploadChecksums {
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");
if (selected != null) {
String trailer = SigV4.single(headers, "x-amz-trailer");
if (algorithm == null && trailer != null) {
Algorithm declared = algorithm(trailer);
if (!selected.equals(declared.name()))
throw new StoreException(400, "InvalidRequest", "Checksum algorithm and trailer must match");
} else if (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);
}
@@ -51,6 +59,10 @@ final class UploadChecksums {
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-crc64nvme" -> new Algorithm("CRC64NVME", header, 8);
case "x-amz-checksum-xxhash64" -> new Algorithm("XXHASH64", header, 8);
case "x-amz-checksum-xxhash3" -> new Algorithm("XXHASH3", header, 8);
case "x-amz-checksum-xxhash128" -> new Algorithm("XXHASH128", header, 16);
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);
@@ -71,6 +83,10 @@ final class UploadChecksums {
return algorithm != null && algorithm.name().equals("SHA256") ? encoded : null;
}
java.util.Map<String, String> metadata() {
return algorithm == null ? java.util.Map.of() : java.util.Map.of(algorithm.header(), encoded);
}
void response(Headers headers) {
if (algorithm != null) headers.set(algorithm.header(), encoded);
}
@@ -91,8 +107,11 @@ final class UploadChecksums {
private final Checksum crc = algorithm == null ? null : switch (algorithm.name()) {
case "CRC32" -> new CRC32();
case "CRC32C" -> new CRC32C();
case "CRC64NVME" -> new Crc64Nvme();
default -> null;
};
private final XxHashes xxhash = algorithm != null && algorithm.name().startsWith("XXHASH")
? new XxHashes(algorithm.name()) : null;
private boolean checked;
private VerifiedInput(InputStream input) { super(input); }
@@ -115,6 +134,7 @@ final class UploadChecksums {
if (md5 != null) md5.update(bytes, offset, length);
if (hash != null) hash.update(bytes, offset, length);
if (crc != null) crc.update(bytes, offset, length);
if (xxhash != null) xxhash.update(bytes, offset, length);
}
private void verify() {
@@ -127,9 +147,13 @@ final class UploadChecksums {
byte[] actual;
if (crc != null) {
long value = crc.getValue();
actual = new byte[]{(byte) (value >>> 24), (byte) (value >>> 16),
(byte) (value >>> 8), (byte) value};
actual = new byte[algorithm.length()];
for (int i = actual.length - 1; i >= 0; i--) {
actual[i] = (byte) value;
value >>>= 8;
}
} else if (algorithm.name().equals("MD5")) actual = actualMd5;
else if (xxhash != null) actual = xxhash.digest();
else actual = hash.digest();
if (!MessageDigest.isEqual(expected, actual))
throw new StoreException(400, "BadDigest", algorithm.name() + " checksum mismatch");
+1 -1
View File
@@ -1,7 +1,7 @@
package cloud.lunarsky.store;
final class Version {
static final String VALUE = "0.0.4";
static final String VALUE = "0.0.8";
private Version() {}
}
+32
View File
@@ -0,0 +1,32 @@
package cloud.lunarsky.store;
import com.dynatrace.hash4j.hashing.HashStream64;
import com.dynatrace.hash4j.hashing.HashStream128;
import com.dynatrace.hash4j.hashing.Hashing;
import java.nio.ByteBuffer;
final class XxHashes {
private final HashStream64 stream;
XxHashes(String algorithm) {
stream = switch (algorithm) {
case "XXHASH64" -> Hashing.xxh64().hashStream();
case "XXHASH3" -> Hashing.xxh3_64().hashStream();
case "XXHASH128" -> Hashing.xxh3_128().hashStream();
default -> throw new IllegalArgumentException(algorithm);
};
}
void update(byte[] bytes, int offset, int length) {
stream.putBytes(bytes, offset, length);
}
byte[] digest() {
if (stream instanceof HashStream128 wide) {
var value = wide.get();
return ByteBuffer.allocate(16).putLong(value.getMostSignificantBits())
.putLong(value.getLeastSignificantBits()).array();
}
return ByteBuffer.allocate(8).putLong(stream.getAsLong()).array();
}
}