From 1703cbae4963e2bd35df65478455b2d6d87689b7 Mon Sep 17 00:00:00 2001 From: Caleb Hulbert Date: Fri, 13 Feb 2026 16:03:24 -0500 Subject: [PATCH 1/8] feat: volatile createReadData and IoPolicy --- .../n5/s3/AmazonS3KeyValueAccess.java | 396 ++++++++---------- .../janelia/saalfeldlab/n5/s3/S3IoPolicy.java | 106 +++++ 2 files changed, 273 insertions(+), 229 deletions(-) create mode 100644 src/main/java/org/janelia/saalfeldlab/n5/s3/S3IoPolicy.java diff --git a/src/main/java/org/janelia/saalfeldlab/n5/s3/AmazonS3KeyValueAccess.java b/src/main/java/org/janelia/saalfeldlab/n5/s3/AmazonS3KeyValueAccess.java index a14abf7..24693b7 100644 --- a/src/main/java/org/janelia/saalfeldlab/n5/s3/AmazonS3KeyValueAccess.java +++ b/src/main/java/org/janelia/saalfeldlab/n5/s3/AmazonS3KeyValueAccess.java @@ -25,24 +25,13 @@ */ package org.janelia.saalfeldlab.n5.s3; -import java.io.ByteArrayOutputStream; -import java.io.Closeable; -import java.io.IOException; -import java.io.InputStream; -import java.io.InputStreamReader; -import java.io.OutputStream; -import java.io.OutputStreamWriter; -import java.io.Reader; -import java.io.Writer; +import java.io.*; import java.net.URI; import java.net.URISyntaxException; -import java.nio.channels.NonReadableChannelException; -import java.nio.charset.StandardCharsets; import java.nio.file.Path; import java.util.ArrayList; import java.util.List; import java.util.function.Supplier; -import java.util.stream.Collectors; import org.janelia.saalfeldlab.n5.KeyValueAccess; import org.janelia.saalfeldlab.n5.LockedChannel; @@ -60,9 +49,6 @@ import software.amazon.awssdk.core.sync.ResponseTransformer; import software.amazon.awssdk.services.s3.S3Client; import software.amazon.awssdk.services.s3.model.CreateBucketRequest; -import software.amazon.awssdk.services.s3.model.Delete; -import software.amazon.awssdk.services.s3.model.DeleteObjectRequest; -import software.amazon.awssdk.services.s3.model.DeleteObjectsRequest; import software.amazon.awssdk.services.s3.model.GetObjectRequest; import software.amazon.awssdk.services.s3.model.GetObjectResponse; import software.amazon.awssdk.services.s3.model.HeadObjectRequest; @@ -71,7 +57,6 @@ import software.amazon.awssdk.services.s3.model.ListObjectsV2Response; import software.amazon.awssdk.services.s3.model.NoSuchBucketException; import software.amazon.awssdk.services.s3.model.NoSuchKeyException; -import software.amazon.awssdk.services.s3.model.ObjectIdentifier; import software.amazon.awssdk.services.s3.model.PutObjectRequest; import software.amazon.awssdk.services.s3.model.S3Exception; @@ -80,6 +65,7 @@ public class AmazonS3KeyValueAccess implements KeyValueAccess { private final S3Client s3; private final URI containerURI; private final String bucketName; + private final S3IoPolicy ioPolicy; private final boolean createBucket; private Boolean bucketCheckedAndExists = null; @@ -123,6 +109,8 @@ public AmazonS3KeyValueAccess(final S3Client s3, final URI containerURI, final b this.bucketName = AmazonS3Utils.getS3Bucket(containerURI); this.createBucket = createBucket; + this.ioPolicy = setIoPolicy(); + if (!bucketExists()) { if (createBucket) { s3.createBucket(CreateBucketRequest.builder().bucket(bucketName).build()); @@ -134,6 +122,22 @@ public AmazonS3KeyValueAccess(final S3Client s3, final URI containerURI, final b } } + private S3IoPolicy setIoPolicy() { + + String ioPolicy = System.getProperty("n5.ioPolicy"); + if (ioPolicy == null) + return new S3IoPolicy.EtagMatch(s3, bucketName); + + switch (ioPolicy) { + case "unsafe": + case "atomicFallbackUnsafe": // For S3, this is equivalent ot just Unsafe + return new S3IoPolicy.Unsafe(s3, bucketName); + case "atomic": + default: + return new S3IoPolicy.EtagMatch(s3, bucketName); + } + } + private boolean bucketExists() { return bucketCheckedAndExists = bucketCheckedAndExists != null @@ -277,12 +281,9 @@ private boolean keyExists(final String key) { try { // TODO needs testing. // the exception thrown may depend on permissions - @SuppressWarnings("unused") - HeadObjectResponse headObjectResponse = s3.headObject( - HeadObjectRequest.builder().key(key).bucket(bucketName).build()); - + headObjectRequest(s3, bucketName, key, null); return true; - } catch( NoSuchKeyException e ) { + } catch( N5NoSuchKeyException | NoSuchKeyException e ) { return false; } catch (Throwable e) { throw new N5Exception(e); @@ -312,7 +313,7 @@ private boolean prefixExists(final String prefix) { * @param path the path * @return the path with a trailing slash */ - private static String addTrailingSlash(final String path) { + static String addTrailingSlash(final String path) { return path.endsWith("/") ? path : path + "/"; } @@ -324,7 +325,7 @@ private static String addTrailingSlash(final String path) { * @param path the path * @return the path without the leading slash */ - private static String removeLeadingSlash(final String path) { + static String removeLeadingSlash(final String path) { return path.startsWith("/") ? path.substring(1) : path; } @@ -371,108 +372,31 @@ public boolean isFile(final String normalPath) { public long size(String normalPath) throws N5NoSuchKeyException { final String key = removeLeadingSlash(AmazonS3Utils.getS3Key(normalPath)); - - final HeadObjectRequest headObjectRequest = HeadObjectRequest.builder() - .bucket(bucketName) - .key(key) - .build(); - - final HeadObjectResponse response = rethrowS3Exceptions(() -> s3.headObject(headObjectRequest)); - return response.contentLength(); + return headObjectRequest(s3, bucketName, key, null).contentLength(); } @Override public VolatileReadData createReadData(String normalPath) throws N5Exception.N5IOException { - // TODO locking - return VolatileReadData.from(new S3LazyRead(normalPath)); - } - - @Override - public void write(final String normalPath, final ReadData data) throws N5IOException { - - // TODO locking final String key = AmazonS3Utils.getS3Key(normalPath); - try (final S3ObjectChannel ch = new S3ObjectChannel(removeLeadingSlash(key), false); - final OutputStream os = ch.newOutputStream();) { - data.writeTo(os); - } catch (IOException e) { - throw new N5Exception.N5IOException(e); - } - - } - - private T rethrowS3Exceptions(Supplier action) { - try { - return action.get(); - } catch (final NoSuchKeyException e) { - throw new N5Exception.N5NoSuchKeyException("No such key", e); - } catch (final AwsServiceException e) { - final int statusCode = e.awsErrorDetails().sdkHttpResponse().statusCode(); - if (statusCode == 404 || statusCode == 403) - throw new N5Exception.N5NoSuchKeyException("No such key", e); - throw new N5Exception.N5IOException("S3 Exception", e); - } - } - - private class S3LazyRead implements LazyRead { - - private final String key; - - S3LazyRead(final String normalizedKey) { - this.key = normalizedKey; - } - - private GetObjectRequest createObjectRequest(final String s3Key, long offset, long length) { - - final GetObjectRequest.Builder requestBuilder = GetObjectRequest.builder() - .key(s3Key) - .bucket(bucketName); - - // Only add range header if we're doing a partial read - if (offset > 0 || length > 0) { - // HTTP Range header format: "bytes=start-end" - // If length is 0 or negative, read from offset to end of file - final String range = length > 0 - ? String.format("bytes=%d-%d", offset, offset + length - 1) - : String.format("bytes=%d-", offset); - requestBuilder.range(range); - } - - return requestBuilder.build(); - } - - @Override public ReadData materialize(long offset, long length) throws N5Exception.N5IOException { - - final String s3Key = AmazonS3Utils.getS3Key(key); - final GetObjectRequest request = createObjectRequest(s3Key, offset, length); - final ResponseBytes response = rethrowS3Exceptions(() -> s3.getObject(request, ResponseTransformer.toBytes())); - return ReadData.from(response.asByteArray()); - } - - @Override public long size() throws N5Exception.N5IOException { - - return AmazonS3KeyValueAccess.this.size(key); - } - - @Override - public void close() throws IOException { - // TODO ioPolicy? - } + try { + return ioPolicy.read(key); + } catch (IOException e) { + throw new N5IOException(e); + } } @Override - public LockedChannel lockForReading(final String normalPath) { + public void write(final String normalPath, final ReadData data) throws N5IOException { final String key = AmazonS3Utils.getS3Key(normalPath); - return new S3ObjectChannel(key, true); - } - - @Override - public LockedChannel lockForWriting(final String normalPath) { + final String normalizedKey = removeLeadingSlash(key); - final String key = AmazonS3Utils.getS3Key(normalPath); - return new S3ObjectChannel(removeLeadingSlash(key), false); + try { + ioPolicy.write(normalizedKey, data); + } catch (IOException e) { + throw new N5IOException(e); + } } @Override @@ -559,163 +483,177 @@ public void delete(final String normalPath) { } final String key = removeLeadingSlash(AmazonS3Utils.getS3Key(normalPath)); - if (!key.endsWith("/")) { + try { + ioPolicy.delete(key); + } catch (IOException e) { + throw new N5IOException(e); + } + } - final DeleteObjectRequest deleteRequest = DeleteObjectRequest.builder() + static HeadObjectResponse headObjectRequest(final S3Client s3, final String bucketName, final String key, final String matchEtag) { + HeadObjectRequest.Builder requestBuilder = HeadObjectRequest.builder() .bucket(bucketName) - .key(key) - .build(); + .key(key); - try { - s3.deleteObject(deleteRequest); - } catch (S3Exception e) { - } - } + if (matchEtag != null) + requestBuilder.ifMatch(matchEtag); - final String prefix = addTrailingSlash(key); + final HeadObjectRequest request = requestBuilder.build(); - ListObjectsV2Request listObjectsRequest; - ListObjectsV2Response objectsListing; - listObjectsRequest = ListObjectsV2Request.builder() - .bucket(bucketName) - .prefix(prefix) - .build(); + return rethrowS3Exceptions(() -> s3.headObject(request)); + } - do { - objectsListing = s3.listObjectsV2(listObjectsRequest); - final List objectsToDelete = objectsListing.contents().stream().map( x -> { - return ObjectIdentifier.builder().key(x.key()).build(); - }).collect(Collectors.toList()); + static T rethrowS3Exceptions(Supplier action) { + try { + return action.get(); + } catch (final NoSuchKeyException e) { + throw new N5Exception.N5NoSuchKeyException("No such key", e); + } catch (final AwsServiceException e) { + final int statusCode = e.awsErrorDetails().sdkHttpResponse().statusCode(); + if (statusCode == 404 || statusCode == 403) + throw new N5Exception.N5NoSuchKeyException("No such key", e); + if (statusCode == 412) + throw new N5ConcurrentModificationException("eTag has changed since initial request", e); + throw new N5Exception.N5IOException("S3 Exception", e); + } + } - if (!objectsToDelete.isEmpty()) { - final DeleteObjectsRequest deleteRequest = DeleteObjectsRequest.builder() - .bucket(bucketName) - .delete( Delete.builder().objects(objectsToDelete).build()) - .build(); + //TODO: Move to N5 + static class N5ConcurrentModificationException extends N5Exception { - s3.deleteObjects(deleteRequest); - } + public N5ConcurrentModificationException(String message) { + super(message); + } + + public N5ConcurrentModificationException(String message, Throwable cause) { + super(message, cause); + } - // TODO what about continuation token? + public N5ConcurrentModificationException(Throwable cause) { + super(cause); + } - } while (objectsListing.isTruncated()); } - private class S3ObjectChannel implements LockedChannel { + @Override + @Deprecated + public LockedChannel lockForReading(final String normalPath) { - protected final String path; - final boolean readOnly; - private final ArrayList resources = new ArrayList<>(); + throw new UnsupportedOperationException("Deprecated; use `createReadData`"); + } - protected S3ObjectChannel(final String path, final boolean readOnly) { + @Override + @Deprecated + public LockedChannel lockForWriting(final String normalPath) { - this.path = path; - this.readOnly = readOnly; - } + return new LockedChannel() { - private void checkWritable() { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); - if (readOnly) { - throw new NonReadableChannelException(); + @Override + public Reader newReader() throws N5IOException { + return null; } - } - @Override - public InputStream newInputStream() { - - try { - final GetObjectRequest objectRequest = GetObjectRequest.builder().key(path).bucket(bucketName).build(); - // TODO consider using ResponseTransformer.toBytes - return s3.getObject(objectRequest, ResponseTransformer.toInputStream()); - } catch (final S3Exception e) { - // TODO figure out how to determine if the error is because the key does not exist - if (e.statusCode() == 404 || e.statusCode() == 403) - throw new N5Exception.N5NoSuchKeyException("No such key", e); - else - throw new N5Exception.N5IOException(e); + @Override + public InputStream newInputStream() throws N5IOException { + return null; } - } - - @Override - public Reader newReader() { - final InputStreamReader reader = new InputStreamReader(newInputStream(), StandardCharsets.UTF_8); - synchronized (resources) { - resources.add(reader); + @Override + public Writer newWriter() throws N5IOException { + return new BufferedWriter(new OutputStreamWriter(baos)); } - return reader; - } - @Override - public OutputStream newOutputStream() { + @Override + public OutputStream newOutputStream() throws N5IOException { + return baos; + } - checkWritable(); - final S3OutputStream s3Out = new S3OutputStream(); - synchronized (resources) { - resources.add(s3Out); + @Override + public void close() throws IOException { + ReadData readData = ReadData.from(baos.toByteArray()); + write(normalPath, readData); } - return s3Out; - } + }; + } - @Override - public Writer newWriter() { + static class S3LazyRead implements LazyRead { - checkWritable(); - final OutputStreamWriter writer = new OutputStreamWriter(newOutputStream(), StandardCharsets.UTF_8); - synchronized (resources) { - resources.add(writer); - } - return writer; + private final String s3Key; + private final boolean verifyEtag; + private final S3Client s3; + private final String bucketName; + private String eTag = null; + + + S3LazyRead(final S3Client s3, final String bucketName, final String s3Key, final boolean verifyEtag) { + this.s3 = s3; + this.bucketName = bucketName; + this.s3Key = s3Key; + this.verifyEtag = verifyEtag; } - @Override - public void close() throws IOException { + private String getEtag(final String s3Key) { + HeadObjectRequest headRequest = HeadObjectRequest.builder() + .bucket(bucketName) + .key(s3Key) + .build(); - synchronized (resources) { - for (final Closeable resource : resources) - resource.close(); - resources.clear(); - } + return s3.headObject(headRequest).eTag(); } - final class S3OutputStream extends OutputStream { + private GetObjectRequest createObjectRequest(final String s3Key, long offset, long length) { - private final ByteArrayOutputStream buf = new ByteArrayOutputStream(); - private boolean closed = false; + final GetObjectRequest.Builder requestBuilder = GetObjectRequest.builder() + .key(s3Key) + .bucket(bucketName); - @Override - public void write(final byte[] b, final int off, final int len) { + // Only add range header if we're doing a partial read + if (offset > 0 || length > 0) { + // HTTP Range header format: "bytes=start-end" + // If length is 0 or negative, read from offset to end of file + final String range = length > 0 + ? String.format("bytes=%d-%d", offset, offset + length - 1) + : String.format("bytes=%d-", offset); + requestBuilder.range(range); + } - buf.write(b, off, len); + if (verifyEtag) { + if (eTag == null) + eTag = getEtag(s3Key); + requestBuilder.ifMatch(eTag); } - @Override - public void write(final int b) { + return requestBuilder.build(); + } - buf.write(b); - } + @Override public ReadData materialize(long offset, long length) throws N5Exception.N5IOException { - @Override - public synchronized void close() throws IOException { - - if (!closed) { - closed = true; - PutObjectRequest putOb = PutObjectRequest.builder() - .bucket(bucketName) - .key(path) - .build(); - - try { - s3.putObject(putOb, RequestBody.fromBytes(buf.toByteArray())); - } catch (S3Exception e) { - e.printStackTrace(); - } - buf.close(); - } - } + final ResponseBytes response = rethrowS3Exceptions(() -> { + final GetObjectRequest request = createObjectRequest(s3Key, offset, length); + return s3.getObject(request, ResponseTransformer.toBytes()); + }); + return ReadData.from(response.asByteArray()); } - } + @Override public long size() throws N5Exception.N5IOException { + + if (!verifyEtag) + eTag = null; + + final HeadObjectResponse response = headObjectRequest(s3, bucketName, s3Key, eTag); + + if (eTag == null && verifyEtag) + eTag = response.eTag(); + return response.contentLength(); + } + + @Override + public void close() { + eTag = null; + } + } } diff --git a/src/main/java/org/janelia/saalfeldlab/n5/s3/S3IoPolicy.java b/src/main/java/org/janelia/saalfeldlab/n5/s3/S3IoPolicy.java new file mode 100644 index 0000000..2a777ec --- /dev/null +++ b/src/main/java/org/janelia/saalfeldlab/n5/s3/S3IoPolicy.java @@ -0,0 +1,106 @@ +package org.janelia.saalfeldlab.n5.s3; + +import org.janelia.saalfeldlab.n5.IoPolicy; +import org.janelia.saalfeldlab.n5.readdata.ReadData; +import org.janelia.saalfeldlab.n5.readdata.VolatileReadData; +import software.amazon.awssdk.core.sync.RequestBody; +import software.amazon.awssdk.services.s3.S3Client; +import software.amazon.awssdk.services.s3.model.*; + +import java.io.IOException; +import java.util.List; +import java.util.stream.Collectors; + +import static org.janelia.saalfeldlab.n5.s3.AmazonS3KeyValueAccess.addTrailingSlash; + +public interface S3IoPolicy extends IoPolicy { + + class Unsafe implements S3IoPolicy { + + protected final S3Client s3; + protected final String bucketName; + + public Unsafe(S3Client s3, String bucketName) { + this.s3 = s3; + this.bucketName = bucketName; + } + + @Override + public void write(String key, ReadData readData) { + + final PutObjectRequest putRequest = PutObjectRequest.builder() + .bucket(bucketName) + .key(key) + .build(); + + try { + s3.putObject(putRequest, RequestBody.fromByteBuffer(readData.toByteBuffer())); + } catch (S3Exception e) { + e.printStackTrace(); + } + } + + @Override + public VolatileReadData read(String key) { + return VolatileReadData.from(new AmazonS3KeyValueAccess.S3LazyRead(s3, bucketName, key, false)); + } + + @Override + public void delete(String key) { + if (!key.endsWith("/")) { + + final DeleteObjectRequest deleteRequest = DeleteObjectRequest.builder() + .bucket(bucketName) + .key(key) + .build(); + + try { + s3.deleteObject(deleteRequest); + } catch (S3Exception e) { + } + } + + final String prefix = addTrailingSlash(key); + + ListObjectsV2Request listObjectsRequest; + ListObjectsV2Response objectsListing; + listObjectsRequest = ListObjectsV2Request.builder() + .bucket(bucketName) + .prefix(prefix) + .build(); + + do { + objectsListing = s3.listObjectsV2(listObjectsRequest); + final List objectsToDelete = objectsListing.contents().stream().map(x -> { + return ObjectIdentifier.builder().key(x.key()).build(); + }).collect(Collectors.toList()); + + if (!objectsToDelete.isEmpty()) { + final DeleteObjectsRequest deleteRequest = DeleteObjectsRequest.builder() + .bucket(bucketName) + .delete(Delete.builder().objects(objectsToDelete).build()) + .build(); + + s3.deleteObjects(deleteRequest); + } + + // TODO what about continuation token? + + } while (objectsListing.isTruncated()); + } + } + + class EtagMatch extends Unsafe { + + public EtagMatch(S3Client s3, String bucketName) { + super(s3, bucketName); + } + + @Override + public VolatileReadData read(String key) { + return VolatileReadData.from(new AmazonS3KeyValueAccess.S3LazyRead(s3, bucketName, key, true)); + } + } + + ; +} From d3d6e822aa952b65ee5c3823a3c1b384e63b24c4 Mon Sep 17 00:00:00 2001 From: Caleb Hulbert Date: Fri, 13 Feb 2026 16:06:23 -0500 Subject: [PATCH 2/8] chore: fix build version --- pom.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pom.xml b/pom.xml index 1f6befc..357a678 100644 --- a/pom.xml +++ b/pom.xml @@ -11,7 +11,7 @@ org.janelia.saalfeldlab n5-aws-s3 - 4.4.0-alpha-6-SNAPSHOT + 4.4.0-alpha-7-SNAPSHOT N5 AWS S3 N5 library implementation using Amazon Web Services S3 backend. From 80b5cdd1de1948fe950c9a131dc6269b6dbefd8c Mon Sep 17 00:00:00 2001 From: Caleb Hulbert Date: Tue, 17 Feb 2026 11:27:51 -0500 Subject: [PATCH 3/8] perf: only request a single bucket for the endpoint check Signed-off-by: Caleb Hulbert --- .../org/janelia/saalfeldlab/n5/s3/AmazonS3Utils.java | 10 +++------- 1 file changed, 3 insertions(+), 7 deletions(-) diff --git a/src/main/java/org/janelia/saalfeldlab/n5/s3/AmazonS3Utils.java b/src/main/java/org/janelia/saalfeldlab/n5/s3/AmazonS3Utils.java index cf393e3..16004b9 100644 --- a/src/main/java/org/janelia/saalfeldlab/n5/s3/AmazonS3Utils.java +++ b/src/main/java/org/janelia/saalfeldlab/n5/s3/AmazonS3Utils.java @@ -16,12 +16,7 @@ import software.amazon.awssdk.services.s3.S3ClientBuilder; import software.amazon.awssdk.services.s3.S3Uri; import software.amazon.awssdk.services.s3.S3Utilities; -import software.amazon.awssdk.services.s3.model.DeleteBucketRequest; -import software.amazon.awssdk.services.s3.model.DeleteObjectRequest; -import software.amazon.awssdk.services.s3.model.ListObjectsV2Request; -import software.amazon.awssdk.services.s3.model.ListObjectsV2Response; -import software.amazon.awssdk.services.s3.model.S3Exception; -import software.amazon.awssdk.services.s3.model.S3Object; +import software.amazon.awssdk.services.s3.model.*; import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider; import software.amazon.awssdk.auth.credentials.AwsCredentials; @@ -266,8 +261,9 @@ public static S3Client createS3(final String uri, @Nullable final Consumer Date: Thu, 19 Feb 2026 14:53:35 -0500 Subject: [PATCH 4/8] refactor: move S3LazyRead to S3IoPolicy Signed-off-by: Caleb Hulbert --- pom.xml | 2 +- .../n5/s3/AmazonS3KeyValueAccess.java | 163 +----------------- .../janelia/saalfeldlab/n5/s3/S3IoPolicy.java | 78 ++++++++- 3 files changed, 79 insertions(+), 164 deletions(-) diff --git a/pom.xml b/pom.xml index 357a678..75a181c 100644 --- a/pom.xml +++ b/pom.xml @@ -117,7 +117,7 @@ sign,deploy-to-scijava - 4.0.0-alpha-9 + 4.0.0-alpha-9-SNAPSHOT 2.2.2 diff --git a/src/main/java/org/janelia/saalfeldlab/n5/s3/AmazonS3KeyValueAccess.java b/src/main/java/org/janelia/saalfeldlab/n5/s3/AmazonS3KeyValueAccess.java index 24693b7..4078df9 100644 --- a/src/main/java/org/janelia/saalfeldlab/n5/s3/AmazonS3KeyValueAccess.java +++ b/src/main/java/org/janelia/saalfeldlab/n5/s3/AmazonS3KeyValueAccess.java @@ -34,31 +34,17 @@ import java.util.function.Supplier; import org.janelia.saalfeldlab.n5.KeyValueAccess; -import org.janelia.saalfeldlab.n5.LockedChannel; import org.janelia.saalfeldlab.n5.N5Exception; import org.janelia.saalfeldlab.n5.N5Exception.N5IOException; import org.janelia.saalfeldlab.n5.N5Exception.N5NoSuchKeyException; import org.janelia.saalfeldlab.n5.N5URI; -import org.janelia.saalfeldlab.n5.readdata.LazyRead; import org.janelia.saalfeldlab.n5.readdata.ReadData; import org.janelia.saalfeldlab.n5.readdata.VolatileReadData; import software.amazon.awssdk.awscore.exception.AwsServiceException; -import software.amazon.awssdk.core.ResponseBytes; import software.amazon.awssdk.core.sync.RequestBody; -import software.amazon.awssdk.core.sync.ResponseTransformer; import software.amazon.awssdk.services.s3.S3Client; -import software.amazon.awssdk.services.s3.model.CreateBucketRequest; -import software.amazon.awssdk.services.s3.model.GetObjectRequest; -import software.amazon.awssdk.services.s3.model.GetObjectResponse; -import software.amazon.awssdk.services.s3.model.HeadObjectRequest; -import software.amazon.awssdk.services.s3.model.HeadObjectResponse; -import software.amazon.awssdk.services.s3.model.ListObjectsV2Request; -import software.amazon.awssdk.services.s3.model.ListObjectsV2Response; -import software.amazon.awssdk.services.s3.model.NoSuchBucketException; -import software.amazon.awssdk.services.s3.model.NoSuchKeyException; -import software.amazon.awssdk.services.s3.model.PutObjectRequest; -import software.amazon.awssdk.services.s3.model.S3Exception; +import software.amazon.awssdk.services.s3.model.*; public class AmazonS3KeyValueAccess implements KeyValueAccess { @@ -280,7 +266,7 @@ private boolean keyExists(final String key) { try { // TODO needs testing. - // the exception thrown may depend on permissions + // the exception thrown may depend on permissions headObjectRequest(s3, bucketName, key, null); return true; } catch( N5NoSuchKeyException | NoSuchKeyException e ) { @@ -490,7 +476,7 @@ public void delete(final String normalPath) { } } - static HeadObjectResponse headObjectRequest(final S3Client s3, final String bucketName, final String key, final String matchEtag) { + public static HeadObjectResponse headObjectRequest(final S3Client s3, final String bucketName, final String key, final String matchEtag) { HeadObjectRequest.Builder requestBuilder = HeadObjectRequest.builder() .bucket(bucketName) .key(key); @@ -503,7 +489,7 @@ static HeadObjectResponse headObjectRequest(final S3Client s3, final String buck return rethrowS3Exceptions(() -> s3.headObject(request)); } - static T rethrowS3Exceptions(Supplier action) { + public static T rethrowS3Exceptions(Supplier action) { try { return action.get(); } catch (final NoSuchKeyException e) { @@ -513,147 +499,8 @@ static T rethrowS3Exceptions(Supplier action) { if (statusCode == 404 || statusCode == 403) throw new N5Exception.N5NoSuchKeyException("No such key", e); if (statusCode == 412) - throw new N5ConcurrentModificationException("eTag has changed since initial request", e); + throw new N5Exception.N5ConcurrentModificationException("eTag has changed since initial request", e); throw new N5Exception.N5IOException("S3 Exception", e); } } - - //TODO: Move to N5 - static class N5ConcurrentModificationException extends N5Exception { - - public N5ConcurrentModificationException(String message) { - super(message); - } - - public N5ConcurrentModificationException(String message, Throwable cause) { - super(message, cause); - } - - public N5ConcurrentModificationException(Throwable cause) { - super(cause); - } - - } - - @Override - @Deprecated - public LockedChannel lockForReading(final String normalPath) { - - throw new UnsupportedOperationException("Deprecated; use `createReadData`"); - } - - @Override - @Deprecated - public LockedChannel lockForWriting(final String normalPath) { - - return new LockedChannel() { - - ByteArrayOutputStream baos = new ByteArrayOutputStream(); - - @Override - public Reader newReader() throws N5IOException { - return null; - } - - @Override - public InputStream newInputStream() throws N5IOException { - return null; - } - - @Override - public Writer newWriter() throws N5IOException { - return new BufferedWriter(new OutputStreamWriter(baos)); - } - - @Override - public OutputStream newOutputStream() throws N5IOException { - return baos; - } - - @Override - public void close() throws IOException { - ReadData readData = ReadData.from(baos.toByteArray()); - write(normalPath, readData); - } - }; - } - - static class S3LazyRead implements LazyRead { - - private final String s3Key; - private final boolean verifyEtag; - private final S3Client s3; - private final String bucketName; - private String eTag = null; - - - S3LazyRead(final S3Client s3, final String bucketName, final String s3Key, final boolean verifyEtag) { - this.s3 = s3; - this.bucketName = bucketName; - this.s3Key = s3Key; - this.verifyEtag = verifyEtag; - } - - private String getEtag(final String s3Key) { - HeadObjectRequest headRequest = HeadObjectRequest.builder() - .bucket(bucketName) - .key(s3Key) - .build(); - - return s3.headObject(headRequest).eTag(); - } - - private GetObjectRequest createObjectRequest(final String s3Key, long offset, long length) { - - - final GetObjectRequest.Builder requestBuilder = GetObjectRequest.builder() - .key(s3Key) - .bucket(bucketName); - - // Only add range header if we're doing a partial read - if (offset > 0 || length > 0) { - // HTTP Range header format: "bytes=start-end" - // If length is 0 or negative, read from offset to end of file - final String range = length > 0 - ? String.format("bytes=%d-%d", offset, offset + length - 1) - : String.format("bytes=%d-", offset); - requestBuilder.range(range); - } - - if (verifyEtag) { - if (eTag == null) - eTag = getEtag(s3Key); - requestBuilder.ifMatch(eTag); - } - - return requestBuilder.build(); - } - - @Override public ReadData materialize(long offset, long length) throws N5Exception.N5IOException { - - final ResponseBytes response = rethrowS3Exceptions(() -> { - final GetObjectRequest request = createObjectRequest(s3Key, offset, length); - return s3.getObject(request, ResponseTransformer.toBytes()); - }); - return ReadData.from(response.asByteArray()); - } - - @Override public long size() throws N5Exception.N5IOException { - - if (!verifyEtag) - eTag = null; - - final HeadObjectResponse response = headObjectRequest(s3, bucketName, s3Key, eTag); - - if (eTag == null && verifyEtag) - eTag = response.eTag(); - - return response.contentLength(); - } - - @Override - public void close() { - eTag = null; - } - } } diff --git a/src/main/java/org/janelia/saalfeldlab/n5/s3/S3IoPolicy.java b/src/main/java/org/janelia/saalfeldlab/n5/s3/S3IoPolicy.java index 2a777ec..12ccf25 100644 --- a/src/main/java/org/janelia/saalfeldlab/n5/s3/S3IoPolicy.java +++ b/src/main/java/org/janelia/saalfeldlab/n5/s3/S3IoPolicy.java @@ -1,17 +1,20 @@ package org.janelia.saalfeldlab.n5.s3; import org.janelia.saalfeldlab.n5.IoPolicy; +import org.janelia.saalfeldlab.n5.N5Exception; +import org.janelia.saalfeldlab.n5.readdata.LazyRead; import org.janelia.saalfeldlab.n5.readdata.ReadData; import org.janelia.saalfeldlab.n5.readdata.VolatileReadData; +import software.amazon.awssdk.core.ResponseBytes; import software.amazon.awssdk.core.sync.RequestBody; +import software.amazon.awssdk.core.sync.ResponseTransformer; import software.amazon.awssdk.services.s3.S3Client; import software.amazon.awssdk.services.s3.model.*; -import java.io.IOException; import java.util.List; import java.util.stream.Collectors; -import static org.janelia.saalfeldlab.n5.s3.AmazonS3KeyValueAccess.addTrailingSlash; +import static org.janelia.saalfeldlab.n5.s3.AmazonS3KeyValueAccess.*; public interface S3IoPolicy extends IoPolicy { @@ -42,7 +45,7 @@ public void write(String key, ReadData readData) { @Override public VolatileReadData read(String key) { - return VolatileReadData.from(new AmazonS3KeyValueAccess.S3LazyRead(s3, bucketName, key, false)); + return VolatileReadData.from(new S3LazyRead(s3, bucketName, key, false)); } @Override @@ -98,9 +101,74 @@ public EtagMatch(S3Client s3, String bucketName) { @Override public VolatileReadData read(String key) { - return VolatileReadData.from(new AmazonS3KeyValueAccess.S3LazyRead(s3, bucketName, key, true)); + return VolatileReadData.from(new S3LazyRead(s3, bucketName, key, true)); } } - ; + class S3LazyRead implements LazyRead { + + private final String s3Key; + private final boolean verifyEtag; + private final S3Client s3; + private final String bucketName; + private String eTag = null; + + + S3LazyRead(final S3Client s3, final String bucketName, final String s3Key, final boolean verifyEtag) { + this.s3 = s3; + this.bucketName = bucketName; + this.s3Key = s3Key; + this.verifyEtag = verifyEtag; + } + + private GetObjectRequest createObjectRequest(final String s3Key, long offset, long length) { + + + final GetObjectRequest.Builder requestBuilder = GetObjectRequest.builder() + .key(s3Key) + .bucket(bucketName); + + // Only add range header if we're doing a partial read + if (offset > 0 || length > 0) { + // HTTP Range header format: "bytes=start-end" + // If length is 0 or negative, read from offset to end of file + final String range = length > 0 + ? String.format("bytes=%d-%d", offset, offset + length - 1) + : String.format("bytes=%d-", offset); + requestBuilder.range(range); + } + + if (verifyEtag && eTag != null) + requestBuilder.ifMatch(eTag); + + return requestBuilder.build(); + } + + @Override public ReadData materialize(long offset, long length) throws N5Exception.N5IOException { + + final ResponseBytes response = rethrowS3Exceptions(() -> { + final GetObjectRequest request = createObjectRequest(s3Key, offset, length); + ResponseBytes responseBytes = s3.getObject(request, ResponseTransformer.toBytes()); + if (verifyEtag && eTag == null) + eTag = responseBytes.response().eTag(); + return responseBytes; + }); + return ReadData.from(response.asByteArray()); + } + + @Override public long size() throws N5Exception.N5IOException { + + final HeadObjectResponse response = headObjectRequest(s3, bucketName, s3Key, eTag); + + if (verifyEtag && eTag == null) + eTag = response.eTag(); + + return response.contentLength(); + } + + @Override + public void close() { + eTag = null; + } + } } From 0e2dc7d9d15b33b08ea2ed21bd17ddf74afbcab9 Mon Sep 17 00:00:00 2001 From: Caleb Hulbert Date: Mon, 23 Feb 2026 14:52:57 -0500 Subject: [PATCH 5/8] ci: combine main/pr and run against development also --- .github/workflows/build-main.yml | 7 ++++++- .github/workflows/build-pr.yml | 30 ------------------------------ 2 files changed, 6 insertions(+), 31 deletions(-) delete mode 100644 .github/workflows/build-pr.yml diff --git a/.github/workflows/build-main.yml b/.github/workflows/build-main.yml index cd627b2..b1eb9f1 100644 --- a/.github/workflows/build-main.yml +++ b/.github/workflows/build-main.yml @@ -4,8 +4,13 @@ on: push: branches: - master + - development tags: - "*-[0-9]+.*" + pull_request: + branches: + - master + - development jobs: build: @@ -35,4 +40,4 @@ jobs: MAVEN_PASS: ${{ secrets.MAVEN_PASS }} OSSRH_PASS: ${{ secrets.OSSRH_PASS }} SIGNING_ASC: ${{ secrets.SIGNING_ASC }} - AWS_REGION: us-east-1 + AWS_REGION: us-east-1 \ No newline at end of file diff --git a/.github/workflows/build-pr.yml b/.github/workflows/build-pr.yml deleted file mode 100644 index 45882a4..0000000 --- a/.github/workflows/build-pr.yml +++ /dev/null @@ -1,30 +0,0 @@ -name: build PR - -on: - pull_request: - branches: - - master - -jobs: - build: - runs-on: ubuntu-latest - - steps: - - uses: actions/checkout@v4 - - name: Start MinIO Server - uses: comfuture/minio-action@v1 - with: - access_key: minioadmin - secret_key: minioadmin - - name: Set up Java - uses: actions/setup-java@v5 - with: - java-version: '8' - distribution: 'zulu' - cache: 'maven' - - name: Set up CI environment - run: .github/setup.sh - - name: Execute the build - run: .github/build.sh - env: - AWS_REGION: us-east-1 From 4d78784c8eee78802e0ea9b19b3d572b83af9700 Mon Sep 17 00:00:00 2001 From: John Bogovic Date: Tue, 24 Feb 2026 14:10:45 -0500 Subject: [PATCH 6/8] test: S3 IoPolicy --- .../n5/s3/backend/BackendIoPolicyTests.java | 112 ++++++++++++++++++ 1 file changed, 112 insertions(+) create mode 100644 src/test/java/org/janelia/saalfeldlab/n5/s3/backend/BackendIoPolicyTests.java diff --git a/src/test/java/org/janelia/saalfeldlab/n5/s3/backend/BackendIoPolicyTests.java b/src/test/java/org/janelia/saalfeldlab/n5/s3/backend/BackendIoPolicyTests.java new file mode 100644 index 0000000..2650428 --- /dev/null +++ b/src/test/java/org/janelia/saalfeldlab/n5/s3/backend/BackendIoPolicyTests.java @@ -0,0 +1,112 @@ +package org.janelia.saalfeldlab.n5.s3.backend; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertThrows; + +import java.io.IOException; +import java.net.URI; +import java.security.SecureRandom; + +import org.janelia.saalfeldlab.n5.N5Exception; +import org.janelia.saalfeldlab.n5.readdata.ReadData; +import org.janelia.saalfeldlab.n5.readdata.VolatileReadData; +import org.janelia.saalfeldlab.n5.s3.AmazonS3KeyValueAccess; +import org.janelia.saalfeldlab.n5.s3.S3IoPolicy; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; + +import software.amazon.awssdk.services.s3.S3Client; + + +public class BackendIoPolicyTests { + + private static final SecureRandom random = new SecureRandom(); + static final String UNSAFE_KEY = "unsafe/obj"; + static final String ETAG_KEY = "etag/obj"; + + protected static S3Client s3; + protected static AmazonS3KeyValueAccess kva; + protected static String bucketName; + + @BeforeClass + public static void setup() { + + s3 = BackendS3Factory.getOrCreateS3(); + bucketName = "n5-test-" + Long.toUnsignedString(random.nextLong()); + kva = new AmazonS3KeyValueAccess(s3, URI.create("s3://" + bucketName), true); + } + + @AfterClass + public static void teardown() { + + if (kva != null) { + kva.delete("/"); // delete the bucket and everything in it + } + } + + @Test + public void testUnsafe() throws IOException { + + final S3IoPolicy.Unsafe policy = new S3IoPolicy.Unsafe(s3, bucketName); + final byte[] data = {0, 1, 2, 3, 4}; + final byte[] data2 = {5, 6, 7}; + + // write and full read roundtrip + policy.write(UNSAFE_KEY, ReadData.from(data)); + + try (VolatileReadData result = policy.read(UNSAFE_KEY)) { + assertArrayEquals(data, result.allBytes()); + } + + // concurrent modification: + // overwrite the blob, then attempt to materialize, shoud succeed + try (VolatileReadData vrd = policy.read(UNSAFE_KEY)) { + + // should not store the generation + vrd.requireLength(); + + // overwrite to new generation + policy.write(UNSAFE_KEY, ReadData.from(data2)); + + // ensure fetching the data gets the new data and does not error + assertArrayEquals(data2, vrd.allBytes()); + } + } + + @Test + public void testEtagMatch() throws IOException { + + final S3IoPolicy.EtagMatch policy = new S3IoPolicy.EtagMatch(s3, bucketName); + final byte[] data1 = {0, 1, 2, 3, 4}; + final byte[] data2 = {5, 6, 7}; + + // write and full read roundtrip + policy.write(ETAG_KEY, ReadData.from(data1)); + try (VolatileReadData result = policy.read(ETAG_KEY)) { + assertArrayEquals(data1, result.allBytes()); + } + + // after close(), a new read sees updated content + policy.write(ETAG_KEY, ReadData.from(data2)); + try (VolatileReadData result = policy.read(ETAG_KEY)) { + assertArrayEquals(data2, result.allBytes()); + } + + // concurrent modification: pin the generation via requireLength(), + // overwrite the blob, then attempt to materialize + policy.write(ETAG_KEY, ReadData.from(data1)); + try (VolatileReadData vrd = policy.read(ETAG_KEY)) { + + // sets the generation of vrd + vrd.requireLength(); + + // overwrite to new generation + policy.write(ETAG_KEY, ReadData.from(data2)); + + // ensure fetching the data + assertThrows(N5Exception.N5ConcurrentModificationException.class, vrd::allBytes); + } + } + +} From 81692f422154b5550325650299bf657cbe1c1f32 Mon Sep 17 00:00:00 2001 From: Caleb Hulbert Date: Wed, 25 Feb 2026 11:23:55 -0500 Subject: [PATCH 7/8] feat: ioPolicy is instance settable, permissive defaults to strict Signed-off-by: Caleb Hulbert --- .../saalfeldlab/n5/s3/AmazonS3KeyValueAccess.java | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/src/main/java/org/janelia/saalfeldlab/n5/s3/AmazonS3KeyValueAccess.java b/src/main/java/org/janelia/saalfeldlab/n5/s3/AmazonS3KeyValueAccess.java index 4078df9..b97c9d9 100644 --- a/src/main/java/org/janelia/saalfeldlab/n5/s3/AmazonS3KeyValueAccess.java +++ b/src/main/java/org/janelia/saalfeldlab/n5/s3/AmazonS3KeyValueAccess.java @@ -51,7 +51,7 @@ public class AmazonS3KeyValueAccess implements KeyValueAccess { private final S3Client s3; private final URI containerURI; private final String bucketName; - private final S3IoPolicy ioPolicy; + private S3IoPolicy ioPolicy; private final boolean createBucket; private Boolean bucketCheckedAndExists = null; @@ -108,6 +108,10 @@ public AmazonS3KeyValueAccess(final S3Client s3, final URI containerURI, final b } } + public void setIoPolicy(S3IoPolicy ioPolicy) { + this.ioPolicy = ioPolicy; + } + private S3IoPolicy setIoPolicy() { String ioPolicy = System.getProperty("n5.ioPolicy"); @@ -116,9 +120,9 @@ private S3IoPolicy setIoPolicy() { switch (ioPolicy) { case "unsafe": - case "atomicFallbackUnsafe": // For S3, this is equivalent ot just Unsafe return new S3IoPolicy.Unsafe(s3, bucketName); - case "atomic": + case "permissive": // For S3, this is equivalent ot just strict + case "strict": default: return new S3IoPolicy.EtagMatch(s3, bucketName); } From 71d74b85c0274de2f585f1722613cb2dbc95b39f Mon Sep 17 00:00:00 2001 From: Caleb Hulbert Date: Wed, 25 Feb 2026 15:51:19 -0500 Subject: [PATCH 8/8] build: bump n5 version Signed-off-by: Caleb Hulbert --- pom.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pom.xml b/pom.xml index 75a181c..2cf8cd2 100644 --- a/pom.xml +++ b/pom.xml @@ -117,7 +117,7 @@ sign,deploy-to-scijava - 4.0.0-alpha-9-SNAPSHOT + 4.0.0-alpha-10 2.2.2