Investigate "Volatile" Reads for S3

[cmhulbert]

When reading multiple ranges from the same key, it would be nice to know that the underlying data has changed if someone wrote in the meantime. This is what I'm referring to as "volatile". For Filesystems you can Lock, but for S3 there is no such thing. This is one potential solution:

fun rangedReadPinnedToSameObject(
    s3: S3Client,
    bucket: String,
    key: String,
    start: Long,
    endInclusive: Long
): ByteArray {
    val head = s3.headObject(
        HeadObjectRequest.builder().bucket(bucket).key(key).build()
    )
    val etag = head.eTag()

    val req = GetObjectRequest.builder()
        .bucket(bucket)
        .key(key)
        .range("bytes=$start-$endInclusive")
        .ifMatch(etag)              
        .build()

    s3.getObject(req).use { input ->
        return input.readBytes()
    }
}

I believe this would fail if the eTag doesn't match, and throw an error which we could report.

This is untested though, needs more investigation

[cmhulbert]

I was able to verify this can be used to ensure the version of a key is not modified across multiple get requests. On modification the eTag is changed, and ifMatch correctly succeeds before modification, and fails after modification. I'll plan to try and implement this when the IoPolicy interface is available in N5

[cmhulbert]

For reference, here is the code I tested, and the output:

UUID uuid = UUID.randomUUID();
String path = "s3://n5-test-bucket-" + uuid;
URI uri = URI.create(path);
S3Client s3 = AmazonS3Utils.createS3(path);
AmazonS3KeyValueAccess kva = new AmazonS3KeyValueAccess(s3, uri, true);
String key = kva.compose(uri, "test");

try (
		LockedChannel test = kva.lockForWriting(key);
		OutputStream os = test.newOutputStream();
		OutputStreamWriter outputStreamWriter = new OutputStreamWriter(os, StandardCharsets.UTF_8);
) {

	outputStreamWriter.write("before");
	outputStreamWriter.flush();
	System.out.println("Wrote 'before' to key 'test'");
}

System.out.print("Read from 'test':");
new BufferedReader( new InputStreamReader(kva.createReadData(key).inputStream()))
		.lines()
		.forEach(System.out::println);
System.out.println("Success\n");



String s3Key = AmazonS3Utils.getS3Key(key);
HeadObjectRequest headRequest = HeadObjectRequest.builder()
		.bucket(kva.bucketName)
		.key(s3Key)
		.build();

String eTag = s3.headObject(headRequest).eTag();
System.out.println("Current eTag is " + eTag + ".");

GetObjectRequest getObjReq = GetObjectRequest.builder()
		.bucket(kva.bucketName)
		.key(s3Key)
		.ifMatch(eTag)
		.build();


System.out.print("Read from 'test' with eTag(" + eTag + "):");
new BufferedReader( new InputStreamReader(s3.getObject(getObjReq, ResponseTransformer.toInputStream())))
		.lines()
		.forEach(System.out::println);
System.out.println("Success\n");

try (
		LockedChannel test = kva.lockForWriting(key);
		OutputStream os = test.newOutputStream();
		OutputStreamWriter outputStreamWriter = new OutputStreamWriter(os, StandardCharsets.UTF_8);
) {

	outputStreamWriter.write("after");
	outputStreamWriter.flush();
}
System.out.println("Wrote 'after' to key 'test'");

System.out.print("Read from 'test' with eTag(" + eTag + "):");
try {
	new BufferedReader( new InputStreamReader(s3.getObject(getObjReq, ResponseTransformer.toInputStream())))
			.lines()
			.forEach(System.out::println);

	System.err.println("Unexpected Success\n");

} catch (Exception e) {
	System.err.println("Expected Exception: " + e.getMessage());
}

String neweTag = s3.headObject(headRequest).eTag();
System.out.println("New eTag is " + neweTag + ".");

GetObjectRequest newGetObjReq = GetObjectRequest.builder()
		.bucket(kva.bucketName)
		.key(s3Key)
		.ifMatch(neweTag)
		.build();

System.out.print("Read from 'test' with eTag(" + neweTag + "):");
new BufferedReader( new InputStreamReader(s3.getObject(newGetObjReq, ResponseTransformer.toInputStream())))
		.lines()
		.forEach(System.out::println);
System.out.println("Success\n");

output:

Wrote 'before' to key 'test'
Read from 'test':before
Success

Current eTag is "2f44417567bc123bd7c60de8c2a2b444".
Read from 'test' with eTag("2f44417567bc123bd7c60de8c2a2b444"):before
Success

Wrote 'after' to key 'test'
Read from 'test' with eTag("2f44417567bc123bd7c60de8c2a2b444"):New eTag is "632a2406bbcbcd553eec45ac14b40a0a".
Expected Exception: At least one of the pre-conditions you specified did not hold (Service: S3, Status Code: 412, Request ID: 8B8VK1E7Y1JRVHGK, Extended Request ID: DaOCGkleVjk8EQEMR7PyypGitpna9zPt/ksjZ1pNBJeu6NSuLXp82W+6gHD8/1yFIWUHayEOreY=)
Read from 'test' with eTag("632a2406bbcbcd553eec45ac14b40a0a"):after
Success

[cmhulbert]

Behavior is merged since #50