From bde1824490eec0eb9ab40176865c58123ba05efc Mon Sep 17 00:00:00 2001 From: Craig McConomy Date: Sun, 2 Aug 2026 16:11:17 -0400 Subject: [PATCH 1/6] feat(blob): blob lease support for the Azure Functions storage backplane MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implement Lease Blob (comp=lease) — acquire, renew, change, release, break — with real lease state reported by Get Blob/Properties and lease-conditioned enforcement on every blob write path (PutBlob, SetBlobMetadata, DeleteBlob, PutBlock, PutBlockList): - active lease + no x-ms-lease-id -> 412 LeaseIdMissing - active lease + wrong id -> 412 LeaseIdMismatchWithBlobOperation - no active lease + id supplied -> 412 LeaseNotPresentWithBlobOperation - competing acquire -> 409 LeaseAlreadyPresent - renew/change/release id mismatch -> 409 LeaseIdMismatchWithLeaseOperation - lease op with no lease -> 409 LeaseNotPresentWithLeaseOperation - renew after break -> 409 LeaseIsBrokenAndCannotBeRenewed - acquire duration outside -1/15-60 -> 400 InvalidHeaderValue Blob leases are the coordination primitive of the Azure Functions host: WebJobs singleton and timer-trigger locks and Durable Functions partition management are all blob leases. Together with the Table OData error fix (#182), this lets a self-hosted Functions fleet point AzureWebJobsStorage at floci-az for durable/timer/queue-triggered workloads — the storage-binding side of #136. (Emulating those triggers inside floci-az's own hosted-function runtime remains open.) Design: lease state is an in-memory map in BlobLeaseService (an emulator restart equals all leases expiring, matching lease semantics); the BlobLease record is a pure function of (lease, now) so expiry and break timing are unit-testable. Container leases stay unimplemented. Verified: unit suite (BlobLeaseTest, BlobLeaseStateTest) red->green; Java SDK compat (BlobLeaseClient lifecycle, write guards, steal-via- break, change handoff) fails on 0.10.0 and passes here; a .NET Azure.Storage.Blobs harness runs the same sequence against Azurite 3.35.0 and this build with line-for-line identical output. Refs #136 --- .../az/compat/BlobCompatibilityTest.java | 94 +++++++ .../io/floci/az/services/blob/BlobLease.java | 60 +++++ .../az/services/blob/BlobLeaseService.java | 231 ++++++++++++++++++ .../az/services/blob/BlobServiceHandler.java | 56 ++++- .../az/services/BlobCompDispatchTest.java | 2 +- .../io/floci/az/services/BlobLeaseTest.java | 219 +++++++++++++++++ .../az/services/blob/BlobLeaseStateTest.java | 61 +++++ 7 files changed, 717 insertions(+), 6 deletions(-) create mode 100644 src/main/java/io/floci/az/services/blob/BlobLease.java create mode 100644 src/main/java/io/floci/az/services/blob/BlobLeaseService.java create mode 100644 src/test/java/io/floci/az/services/BlobLeaseTest.java create mode 100644 src/test/java/io/floci/az/services/blob/BlobLeaseStateTest.java diff --git a/compatibility-tests/sdk-test-java/src/test/java/io/floci/az/compat/BlobCompatibilityTest.java b/compatibility-tests/sdk-test-java/src/test/java/io/floci/az/compat/BlobCompatibilityTest.java index c070988b..c4c2fc31 100644 --- a/compatibility-tests/sdk-test-java/src/test/java/io/floci/az/compat/BlobCompatibilityTest.java +++ b/compatibility-tests/sdk-test-java/src/test/java/io/floci/az/compat/BlobCompatibilityTest.java @@ -17,6 +17,8 @@ import com.azure.storage.blob.models.LeaseStateType; import com.azure.storage.blob.models.LeaseStatusType; import com.azure.storage.blob.models.ListBlobsOptions; +import com.azure.storage.blob.specialized.BlobLeaseClient; +import com.azure.storage.blob.specialized.BlobLeaseClientBuilder; import com.azure.storage.blob.specialized.BlockBlobClient; import com.azure.core.util.Context; import org.junit.jupiter.api.*; @@ -459,4 +461,96 @@ void getServiceProperties() { assertNotNull(props.getHourMetrics()); assertNotNull(props.getMinuteMetrics()); } + + // --- Leases (the WebJobs/Durable Functions host backplane primitive) --- + + private BlobClient leasedBlob(BlobContainerClient container, String blobName) { + BlobClient blob = container.getBlobClient(blobName); + byte[] content = "lock".getBytes(StandardCharsets.UTF_8); + blob.upload(new java.io.ByteArrayInputStream(content), content.length, true); + return blob; + } + + @Test + @DisplayName("blob lease lifecycle: acquire → renew → release") + void blobLeaseLifecycle() { + String name = containerName(); + BlobContainerClient container = client.createBlobContainer(name); + BlobClient blob = leasedBlob(container, "singleton-lock"); + BlobLeaseClient lease = new BlobLeaseClientBuilder().blobClient(blob).buildClient(); + + String leaseId = lease.acquireLease(-1); + assertNotNull(leaseId); + + BlobProperties props = blob.getProperties(); + assertEquals(LeaseStatusType.LOCKED, props.getLeaseStatus()); + assertEquals(LeaseStateType.LEASED, props.getLeaseState()); + + assertEquals(leaseId, lease.renewLease()); + lease.releaseLease(); + + props = blob.getProperties(); + assertEquals(LeaseStatusType.UNLOCKED, props.getLeaseStatus()); + assertEquals(LeaseStateType.AVAILABLE, props.getLeaseState()); + + client.deleteBlobContainer(name); + } + + @Test + @DisplayName("leased blob: writes require the lease id") + void leasedBlobWriteGuards() { + String name = containerName(); + BlobContainerClient container = client.createBlobContainer(name); + BlobClient blob = leasedBlob(container, "guarded"); + BlobLeaseClient lease = new BlobLeaseClientBuilder().blobClient(blob).buildClient(); + String leaseId = lease.acquireLease(-1); + + BlobStorageException ex = assertThrows(BlobStorageException.class, + () -> blob.setMetadata(Map.of("owner", "nobody"))); + assertEquals(412, ex.getStatusCode()); + assertEquals(BlobErrorCode.LEASE_ID_MISSING, ex.getErrorCode()); + + blob.setMetadataWithResponse(Map.of("owner", "host-a"), + new BlobRequestConditions().setLeaseId(leaseId), null, Context.NONE); + assertEquals("host-a", blob.getProperties().getMetadata().get("owner")); + + lease.releaseLease(); + client.deleteBlobContainer(name); + } + + @Test + @DisplayName("competing acquire fails, then succeeds after break (lease steal)") + void leaseStealViaBreak() { + String name = containerName(); + BlobContainerClient container = client.createBlobContainer(name); + BlobClient blob = leasedBlob(container, "partition-lease"); + BlobLeaseClient first = new BlobLeaseClientBuilder().blobClient(blob).buildClient(); + first.acquireLease(-1); + + BlobLeaseClient second = new BlobLeaseClientBuilder().blobClient(blob).buildClient(); + BlobStorageException ex = assertThrows(BlobStorageException.class, + () -> second.acquireLease(-1)); + assertEquals(409, ex.getStatusCode()); + assertEquals(BlobErrorCode.LEASE_ALREADY_PRESENT, ex.getErrorCode()); + + first.breakLease(); + assertNotNull(second.acquireLease(-1)); + + client.deleteBlobContainer(name); + } + + @Test + @DisplayName("change lease: ownership handoff to a proposed id") + void changeLeaseHandsOff() { + String name = containerName(); + BlobContainerClient container = client.createBlobContainer(name); + BlobClient blob = leasedBlob(container, "handoff"); + BlobLeaseClient lease = new BlobLeaseClientBuilder().blobClient(blob).buildClient(); + lease.acquireLease(-1); + + String proposed = UUID.randomUUID().toString(); + assertEquals(proposed, lease.changeLease(proposed)); + + client.deleteBlobContainer(name); + } } diff --git a/src/main/java/io/floci/az/services/blob/BlobLease.java b/src/main/java/io/floci/az/services/blob/BlobLease.java new file mode 100644 index 00000000..d950859a --- /dev/null +++ b/src/main/java/io/floci/az/services/blob/BlobLease.java @@ -0,0 +1,60 @@ +package io.floci.az.services.blob; + +import java.time.Instant; + +/** + * State of one blob lease. Immutable; transitions return a new instance. + * + *

Time is always passed in by the caller so the state machine stays a pure + * function of (lease, now) — expiry and break-elapse never mutate anything. + * + * @param leaseId current lease id (a GUID) + * @param durationSeconds fixed duration 15–60, or -1 for infinite + * @param expiresAt when a fixed lease lapses; null for infinite leases + * @param breakAt when a broken lease's break period ends; null unless broken + */ +public record BlobLease(String leaseId, int durationSeconds, Instant expiresAt, Instant breakAt) { + + public enum State { LEASED, EXPIRED, BREAKING, BROKEN } + + public static BlobLease acquire(String leaseId, int durationSeconds, Instant now) { + Instant expiresAt = durationSeconds < 0 ? null : now.plusSeconds(durationSeconds); + return new BlobLease(leaseId, durationSeconds, expiresAt, null); + } + + public BlobLease renewed(Instant now) { + return acquire(leaseId, durationSeconds, now); + } + + public BlobLease changed(String newLeaseId) { + return new BlobLease(newLeaseId, durationSeconds, expiresAt, breakAt); + } + + public BlobLease broken(int breakPeriodSeconds, Instant now) { + return new BlobLease(leaseId, durationSeconds, expiresAt, now.plusSeconds(breakPeriodSeconds)); + } + + public State stateAt(Instant now) { + if (breakAt != null) { + return now.isBefore(breakAt) ? State.BREAKING : State.BROKEN; + } + if (expiresAt != null && !now.isBefore(expiresAt)) { + return State.EXPIRED; + } + return State.LEASED; + } + + /** An active lease blocks writes and competing acquires. */ + public boolean activeAt(Instant now) { + State s = stateAt(now); + return s == State.LEASED || s == State.BREAKING; + } + + /** Seconds until an in-progress break completes (the x-ms-lease-time of a Break response). */ + public long remainingBreakSeconds(Instant now) { + if (breakAt == null || !now.isBefore(breakAt)) { + return 0; + } + return java.time.Duration.between(now, breakAt).getSeconds(); + } +} diff --git a/src/main/java/io/floci/az/services/blob/BlobLeaseService.java b/src/main/java/io/floci/az/services/blob/BlobLeaseService.java new file mode 100644 index 00000000..03e4518e --- /dev/null +++ b/src/main/java/io/floci/az/services/blob/BlobLeaseService.java @@ -0,0 +1,231 @@ +package io.floci.az.services.blob; + +import io.floci.az.core.AzureErrorResponse; +import io.floci.az.core.AzureRequest; +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.ws.rs.core.Response; + +import java.time.Instant; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; + +/** + * Blob lease state and the Lease Blob operation (comp=lease). + * + *

Leases are the coordination primitive of the Azure Functions host: + * WebJobs singleton/timer locks and Durable Functions partition management + * are blob leases. Lease state is kept in memory only — like a real lease it + * is transient runtime state, and an emulator restart is equivalent to every + * lease having expired. + */ +@ApplicationScoped +public class BlobLeaseService { + + private final Map leases = new ConcurrentHashMap<>(); + + /** + * Dispatch one Lease Blob call for an existing blob. The caller has already + * resolved the blob and 404s when it is absent; a lease operation never + * modifies the blob, so its ETag/Last-Modified are echoed unchanged. + */ + public synchronized Response handleLeaseOp(AzureRequest request, String blobKey, + String etag, String lastModifiedRfc1123) { + String action = header(request, "x-ms-lease-action"); + Instant now = Instant.now(); + BlobLease lease = leases.get(blobKey); + Response response = switch (action == null ? "" : action.toLowerCase()) { + case "acquire" -> acquire(request, blobKey, lease, now); + case "renew" -> renew(request, blobKey, lease, now); + case "change" -> change(request, blobKey, lease, now); + case "release" -> release(request, blobKey, lease, now); + case "break" -> breakLease(request, blobKey, lease, now); + default -> new AzureErrorResponse("InvalidHeaderValue", + "The value for one of the HTTP headers is not in the correct format.") + .toXmlResponse(400); + }; + if (response.getStatus() >= 400) { + return response; + } + return Response.fromResponse(response) + .header("ETag", etag) + .header("Last-Modified", lastModifiedRfc1123) + .build(); + } + + private Response acquire(AzureRequest request, String blobKey, BlobLease lease, Instant now) { + int duration; + try { + duration = Integer.parseInt(header(request, "x-ms-lease-duration") == null + ? "-1" : header(request, "x-ms-lease-duration")); + } catch (NumberFormatException e) { + duration = 0; + } + if (duration != -1 && (duration < 15 || duration > 60)) { + return new AzureErrorResponse("InvalidHeaderValue", + "The value for one of the HTTP headers is not in the correct format.") + .toXmlResponse(400); + } + String proposed = header(request, "x-ms-proposed-lease-id"); + + if (lease != null && lease.activeAt(now)) { + boolean reacquireSameId = lease.stateAt(now) == BlobLease.State.LEASED + && proposed != null && proposed.equalsIgnoreCase(lease.leaseId()); + if (!reacquireSameId) { + return new AzureErrorResponse("LeaseAlreadyPresent", + "There is already a lease present.").toXmlResponse(409); + } + } + + String leaseId = proposed != null && !proposed.isBlank() ? proposed : UUID.randomUUID().toString(); + leases.put(blobKey, BlobLease.acquire(leaseId, duration, now)); + return Response.status(201) + .header("x-ms-lease-id", leaseId) + .build(); + } + + private Response renew(AzureRequest request, String blobKey, BlobLease lease, Instant now) { + String leaseId = header(request, "x-ms-lease-id"); + if (lease == null) { + return leaseNotPresent(); + } + if (!lease.leaseId().equalsIgnoreCase(leaseId)) { + return new AzureErrorResponse("LeaseIdMismatchWithLeaseOperation", + "The lease ID specified did not match the lease ID for the blob.") + .toXmlResponse(409); + } + BlobLease.State state = lease.stateAt(now); + if (state == BlobLease.State.BREAKING || state == BlobLease.State.BROKEN) { + return new AzureErrorResponse("LeaseIsBrokenAndCannotBeRenewed", + "The lease ID matched, but the lease has been broken explicitly and cannot be renewed.") + .toXmlResponse(409); + } + leases.put(blobKey, lease.renewed(now)); + return Response.ok().header("x-ms-lease-id", lease.leaseId()).build(); + } + + private Response change(AzureRequest request, String blobKey, BlobLease lease, Instant now) { + String leaseId = header(request, "x-ms-lease-id"); + String proposed = header(request, "x-ms-proposed-lease-id"); + if (lease == null || !lease.activeAt(now)) { + return leaseNotPresent(); + } + if (!lease.leaseId().equalsIgnoreCase(leaseId)) { + return new AzureErrorResponse("LeaseIdMismatchWithLeaseOperation", + "The lease ID specified did not match the lease ID for the blob.") + .toXmlResponse(409); + } + BlobLease changed = lease.changed(proposed); + leases.put(blobKey, changed); + return Response.ok().header("x-ms-lease-id", changed.leaseId()).build(); + } + + private Response release(AzureRequest request, String blobKey, BlobLease lease, Instant now) { + String leaseId = header(request, "x-ms-lease-id"); + if (lease == null) { + return leaseNotPresent(); + } + if (!lease.leaseId().equalsIgnoreCase(leaseId)) { + return new AzureErrorResponse("LeaseIdMismatchWithLeaseOperation", + "The lease ID specified did not match the lease ID for the blob.") + .toXmlResponse(409); + } + leases.remove(blobKey); + return Response.ok().build(); + } + + private Response breakLease(AzureRequest request, String blobKey, BlobLease lease, Instant now) { + if (lease == null || !lease.activeAt(now)) { + return leaseNotPresent(); + } + int breakPeriod; + String breakPeriodHeader = header(request, "x-ms-lease-break-period"); + if (breakPeriodHeader != null) { + try { + breakPeriod = Math.max(0, Integer.parseInt(breakPeriodHeader)); + } catch (NumberFormatException e) { + return new AzureErrorResponse("InvalidHeaderValue", + "The value for one of the HTTP headers is not in the correct format.") + .toXmlResponse(400); + } + } else { + // Default: infinite leases break immediately, fixed leases run out their term. + breakPeriod = lease.expiresAt() == null ? 0 + : (int) Math.max(0, java.time.Duration.between(now, lease.expiresAt()).getSeconds()); + } + BlobLease broken = lease.stateAt(now) == BlobLease.State.BREAKING + ? lease : lease.broken(breakPeriod, now); + leases.put(blobKey, broken); + return Response.status(202) + .header("x-ms-lease-time", String.valueOf(broken.remainingBreakSeconds(now))) + .build(); + } + + private static Response leaseNotPresent() { + return new AzureErrorResponse("LeaseNotPresentWithLeaseOperation", + "There is currently no lease on the blob.").toXmlResponse(409); + } + + /** + * Lease guard for write/delete operations on a blob. Returns null when the + * operation may proceed, otherwise the 412 the Blob service contract requires. + */ + public Response validateWrite(AzureRequest request, String blobKey) { + String requestLeaseId = header(request, "x-ms-lease-id"); + BlobLease lease = leases.get(blobKey); + Instant now = Instant.now(); + if (lease != null && lease.activeAt(now)) { + if (requestLeaseId == null) { + return new AzureErrorResponse("LeaseIdMissing", + "There is currently a lease on the blob and no lease ID was specified in the request.") + .toXmlResponse(412); + } + if (!lease.leaseId().equalsIgnoreCase(requestLeaseId)) { + return new AzureErrorResponse("LeaseIdMismatchWithBlobOperation", + "The lease ID specified did not match the lease ID for the blob.") + .toXmlResponse(412); + } + } else if (requestLeaseId != null) { + return new AzureErrorResponse("LeaseNotPresentWithBlobOperation", + "There is currently no lease on the blob.").toXmlResponse(412); + } + return null; + } + + /** Stamp x-ms-lease-status/-state (+ -duration when leased) with the blob's real lease state. */ + public void addLeaseHeaders(Response.ResponseBuilder rb, String blobKey) { + BlobLease lease = leases.get(blobKey); + BlobLease.State state = lease == null ? null : lease.stateAt(Instant.now()); + if (state == null) { + rb.header("x-ms-lease-status", "unlocked").header("x-ms-lease-state", "available"); + return; + } + switch (state) { + case LEASED -> rb.header("x-ms-lease-status", "locked") + .header("x-ms-lease-state", "leased") + .header("x-ms-lease-duration", lease.expiresAt() == null ? "infinite" : "fixed"); + case BREAKING -> rb.header("x-ms-lease-status", "locked") + .header("x-ms-lease-state", "breaking"); + case EXPIRED -> rb.header("x-ms-lease-status", "unlocked") + .header("x-ms-lease-state", "expired"); + case BROKEN -> rb.header("x-ms-lease-status", "unlocked") + .header("x-ms-lease-state", "broken"); + } + } + + public void onBlobDeleted(String blobKey) { + leases.remove(blobKey); + } + + public void onContainerDeleted(String blobKeyPrefix) { + leases.keySet().removeIf(k -> k.startsWith(blobKeyPrefix)); + } + + public void clear() { + leases.clear(); + } + + private static String header(AzureRequest request, String name) { + return request.headers().getHeaderString(name); + } +} diff --git a/src/main/java/io/floci/az/services/blob/BlobServiceHandler.java b/src/main/java/io/floci/az/services/blob/BlobServiceHandler.java index a19b0fc3..bf35b878 100644 --- a/src/main/java/io/floci/az/services/blob/BlobServiceHandler.java +++ b/src/main/java/io/floci/az/services/blob/BlobServiceHandler.java @@ -67,13 +67,17 @@ public class BlobServiceHandler implements AzureServiceHandler, Resettable { private final StorageSasAuthorization sasAuthorization; private final DataLakePathOperations dataLakePathOperations; + private final BlobLeaseService leaseService; + @Inject public BlobServiceHandler(StorageFactory storageFactory, EmulatorConfig config, UserDelegationKeyService userDelegationKeyService, - StorageSasAuthorization sasAuthorization) { + StorageSasAuthorization sasAuthorization, + BlobLeaseService leaseService) { this.config = config; this.userDelegationKeyService = userDelegationKeyService; this.sasAuthorization = sasAuthorization; + this.leaseService = leaseService; this.store = storageFactory.create("blob"); this.dataLakePathOperations = new DataLakePathOperations(store); } @@ -164,7 +168,9 @@ public Response handle(AzureRequest request) { response = notImplemented(); } } else { - if ("PUT".equalsIgnoreCase(method) && "metadata".equals(comp)) { + if ("PUT".equalsIgnoreCase(method) && "lease".equals(comp)) { + response = leaseBlob(request, containerName, blobName); + } else if ("PUT".equalsIgnoreCase(method) && "metadata".equals(comp)) { response = setBlobMetadata(request, containerName, blobName); } else if (("GET".equalsIgnoreCase(method) || "HEAD".equalsIgnoreCase(method)) && "metadata".equals(comp)) { @@ -210,6 +216,17 @@ private Response notImplemented() { .toXmlResponse(501); } + /** PUT /{container}/{blob}?comp=lease — Lease Blob (acquire/renew/change/release/break). */ + private Response leaseBlob(AzureRequest request, String containerName, String blobName) { + Optional object = store.get(objKey(request.accountName(), containerName, blobName)); + if (object.isEmpty()) { + return new AzureErrorResponse("BlobNotFound", "The specified blob does not exist.") + .toXmlResponse(Response.Status.NOT_FOUND.getStatusCode()); + } + return leaseService.handleLeaseOp(request, objKey(request.accountName(), containerName, blobName), + object.get().etag(), RFC1123_DATE_TIME.format(object.get().lastModified())); + } + /** * True only for a genuine PutBlob. * @@ -302,6 +319,7 @@ private Response deleteContainer(AzureRequest request, String containerName) { .filter(k -> k.startsWith(objPrefix) || k.startsWith(blkPrefix)) .toList() .forEach(store::delete); + leaseService.onContainerDeleted(objPrefix); return Response.status(Response.Status.ACCEPTED).build(); } @@ -348,6 +366,11 @@ private Response putBlob(AzureRequest request, String containerName, String blob if (conditionFailure != null) { return conditionFailure; } + Response leaseFailure = leaseService.validateWrite(request, + objKey(request.accountName(), containerName, blobName)); + if (leaseFailure != null) { + return leaseFailure; + } byte[] data = request.bodyStream().readAllBytes(); Map metadata = new HashMap<>(); @@ -435,10 +458,8 @@ private Response getBlob(AzureRequest request, String containerName, String blob .header("Accept-Ranges", "bytes") // Get Blob always reports these. Strict SDK clients (e.g. the Azure SDK for C++) // read x-ms-creation-time and x-ms-server-encrypted unconditionally and throw when - // they are absent. Leases are not modelled, so the values are those of an unleased blob. + // they are absent. .header("x-ms-creation-time", RFC1123_DATE_TIME.format(creationTime(so))) - .header("x-ms-lease-status", "unlocked") - .header("x-ms-lease-state", "available") .header("x-ms-server-encrypted", "true"); for (String header : BLOB_HTTP_PROPERTY_HEADERS.values()) { String value = so.metadata().get(header); @@ -446,6 +467,7 @@ private Response getBlob(AzureRequest request, String containerName, String blob rb.header(header, value); } } + leaseService.addLeaseHeaders(rb, objKey(request.accountName(), containerName, blobName)); if (isRangeRequest) { rb.header("Content-Range", String.format("bytes %d-%d/%d", rangeStart, rangeEnd, totalSize)); } @@ -477,7 +499,13 @@ private Response deleteBlob(AzureRequest request, String containerName, String b if (conditionFailure != null) { return conditionFailure; } + Response leaseFailure = leaseService.validateWrite(request, + objKey(request.accountName(), containerName, blobName)); + if (leaseFailure != null) { + return leaseFailure; + } store.delete(objKey(request.accountName(), containerName, blobName)); + leaseService.onBlobDeleted(objKey(request.accountName(), containerName, blobName)); return Response.status(Response.Status.ACCEPTED).build(); } @@ -521,6 +549,12 @@ private Response setBlobMetadata(AzureRequest request, String containerName, Str return conditionFailure; } + Response leaseFailure = leaseService.validateWrite(request, + objKey(request.accountName(), containerName, blobName)); + if (leaseFailure != null) { + return leaseFailure; + } + StoredObject so = object.get(); Map metadata = new HashMap<>(); so.metadata().forEach((key, value) -> { @@ -689,6 +723,11 @@ private Response putBlock(AzureRequest request, String containerName, String blo "Value for one of the query parameters specified in the request URI is invalid.") .toXmlResponse(400); } + Response leaseFailure = leaseService.validateWrite(request, + objKey(request.accountName(), containerName, blobName)); + if (leaseFailure != null) { + return leaseFailure; + } byte[] data = request.bodyStream().readAllBytes(); store.put(blockStagingKey(request.accountName(), containerName, blobName, blockId), new StoredObject(blockId, data, Map.of("BlockId", blockId), Instant.now(), @@ -719,6 +758,12 @@ private Response putBlockList(AzureRequest request, String containerName, String .toXmlResponse(Response.Status.NOT_FOUND.getStatusCode()); } + Response leaseFailure = leaseService.validateWrite(request, + objKey(request.accountName(), containerName, blobName)); + if (leaseFailure != null) { + return leaseFailure; + } + List blockIds = parseBlockList(request.bodyStream().readAllBytes()); // Resolve every block ID → staged data @@ -945,6 +990,7 @@ private static List parseBlockList(byte[] body) { public void clear() { store.clear(); + leaseService.clear(); } public void ensureContainer(String accountName, String containerName) { diff --git a/src/test/java/io/floci/az/services/BlobCompDispatchTest.java b/src/test/java/io/floci/az/services/BlobCompDispatchTest.java index 8b309c81..9422c85e 100644 --- a/src/test/java/io/floci/az/services/BlobCompDispatchTest.java +++ b/src/test/java/io/floci/az/services/BlobCompDispatchTest.java @@ -56,7 +56,7 @@ private void assertBlobIntact() { */ @ParameterizedTest @ValueSource(strings = { - "lease", "snapshot", "properties", "tier", "tags", + "snapshot", "properties", "tier", "tags", "page", "appendblock", "undelete", "expiry", "seal" }) void unimplementedBlobCompIsNotMistakenForPutBlob(String comp) { diff --git a/src/test/java/io/floci/az/services/BlobLeaseTest.java b/src/test/java/io/floci/az/services/BlobLeaseTest.java new file mode 100644 index 00000000..039d25c5 --- /dev/null +++ b/src/test/java/io/floci/az/services/BlobLeaseTest.java @@ -0,0 +1,219 @@ +package io.floci.az.services; + +import io.quarkus.test.junit.QuarkusTest; +import io.restassured.response.Response; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import static io.restassured.RestAssured.given; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.*; + +/** + * Blob lease operations (comp=lease): the backplane the Azure Functions host + * depends on — WebJobs singleton/timer locks and Durable Functions partition + * leases are all blob leases (see floci-io/floci-az#136). + */ +@QuarkusTest +public class BlobLeaseTest { + + private static final String ACCOUNT = "devstoreaccount1"; + private static final String CONTAINER = "lease-container"; + private static final String BLOB = "lock-blob"; + + @BeforeEach + void reset() { + given().post("/_admin/reset").then().statusCode(204); + given().put("/{account}/{container}?restype=container", ACCOUNT, CONTAINER) + .then().statusCode(201); + given().body("lock").put("/{account}/{container}/{blob}", ACCOUNT, CONTAINER, BLOB) + .then().statusCode(201); + } + + private Response leaseOp(String action, String... headers) { + var spec = given().header("x-ms-lease-action", action); + for (int i = 0; i + 1 < headers.length; i += 2) { + spec = spec.header(headers[i], headers[i + 1]); + } + return spec.put("/{account}/{container}/{blob}?comp=lease", ACCOUNT, CONTAINER, BLOB); + } + + private String acquire() { + Response r = leaseOp("acquire", "x-ms-lease-duration", "-1"); + r.then().statusCode(201); + return r.header("x-ms-lease-id"); + } + + // ── Lifecycle ──────────────────────────────────────────────────────────── + + @Test + void acquireReturnsLeaseIdAndLocksBlob() { + Response r = leaseOp("acquire", "x-ms-lease-duration", "-1"); + r.then().statusCode(201); + assertThat(r.header("x-ms-lease-id"), not(emptyOrNullString())); + + given().head("/{account}/{container}/{blob}", ACCOUNT, CONTAINER, BLOB) + .then() + .header("x-ms-lease-status", equalTo("locked")) + .header("x-ms-lease-state", equalTo("leased")) + .header("x-ms-lease-duration", equalTo("infinite")); + } + + @Test + void acquireWithProposedIdReturnsProposedIdAndIsIdempotent() { + String proposed = "11111111-2222-3333-4444-555555555555"; + leaseOp("acquire", "x-ms-lease-duration", "-1", "x-ms-proposed-lease-id", proposed) + .then().statusCode(201).header("x-ms-lease-id", equalTo(proposed)); + + // Re-acquire with the same proposed id succeeds (the SDK retries this way). + leaseOp("acquire", "x-ms-lease-duration", "-1", "x-ms-proposed-lease-id", proposed) + .then().statusCode(201).header("x-ms-lease-id", equalTo(proposed)); + } + + @Test + void acquireWhenLeasedReturnsLeaseAlreadyPresent() { + acquire(); + leaseOp("acquire", "x-ms-lease-duration", "-1") + .then().statusCode(409).header("x-ms-error-code", equalTo("LeaseAlreadyPresent")); + } + + @Test + void acquireWithInvalidDurationReturnsInvalidHeaderValue() { + leaseOp("acquire", "x-ms-lease-duration", "5") + .then().statusCode(400).header("x-ms-error-code", equalTo("InvalidHeaderValue")); + } + + @Test + void acquireOnMissingBlobReturnsBlobNotFound() { + given().header("x-ms-lease-action", "acquire").header("x-ms-lease-duration", "-1") + .put("/{account}/{container}/absent?comp=lease", ACCOUNT, CONTAINER) + .then().statusCode(404).header("x-ms-error-code", equalTo("BlobNotFound")); + } + + @Test + void renewWithCorrectIdSucceedsWrongIdConflicts() { + String id = acquire(); + leaseOp("renew", "x-ms-lease-id", id) + .then().statusCode(200).header("x-ms-lease-id", equalTo(id)); + leaseOp("renew", "x-ms-lease-id", "99999999-9999-9999-9999-999999999999") + .then().statusCode(409) + .header("x-ms-error-code", equalTo("LeaseIdMismatchWithLeaseOperation")); + } + + @Test + void renewWithoutActiveLeaseReturnsLeaseNotPresent() { + leaseOp("renew", "x-ms-lease-id", "99999999-9999-9999-9999-999999999999") + .then().statusCode(409) + .header("x-ms-error-code", equalTo("LeaseNotPresentWithLeaseOperation")); + } + + @Test + void releaseUnlocksAndAllowsReacquire() { + String id = acquire(); + leaseOp("release", "x-ms-lease-id", id).then().statusCode(200); + + given().head("/{account}/{container}/{blob}", ACCOUNT, CONTAINER, BLOB) + .then() + .header("x-ms-lease-status", equalTo("unlocked")) + .header("x-ms-lease-state", equalTo("available")); + + Response r = leaseOp("acquire", "x-ms-lease-duration", "-1"); + r.then().statusCode(201); + assertThat(r.header("x-ms-lease-id"), not(equalTo(id))); + } + + @Test + void changeLeaseSwapsToProposedId() { + String id = acquire(); + String proposed = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"; + leaseOp("change", "x-ms-lease-id", id, "x-ms-proposed-lease-id", proposed) + .then().statusCode(200).header("x-ms-lease-id", equalTo(proposed)); + + // Old id no longer renews; new id does. + leaseOp("renew", "x-ms-lease-id", id) + .then().statusCode(409) + .header("x-ms-error-code", equalTo("LeaseIdMismatchWithLeaseOperation")); + leaseOp("renew", "x-ms-lease-id", proposed).then().statusCode(200); + } + + @Test + void breakLeaseAllowsReacquireAndBlocksRenew() { + String id = acquire(); + Response broken = leaseOp("break"); + broken.then().statusCode(202); + assertThat(broken.header("x-ms-lease-time"), equalTo("0")); + + leaseOp("renew", "x-ms-lease-id", id) + .then().statusCode(409) + .header("x-ms-error-code", equalTo("LeaseIsBrokenAndCannotBeRenewed")); + + leaseOp("acquire", "x-ms-lease-duration", "-1").then().statusCode(201); + } + + @Test + void breakWithoutActiveLeaseReturnsLeaseNotPresent() { + leaseOp("break") + .then().statusCode(409) + .header("x-ms-error-code", equalTo("LeaseNotPresentWithLeaseOperation")); + } + + // ── Write guards ───────────────────────────────────────────────────────── + + @Test + void writeToLeasedBlobRequiresLeaseId() { + String id = acquire(); + + given().body("update") + .put("/{account}/{container}/{blob}", ACCOUNT, CONTAINER, BLOB) + .then().statusCode(412).header("x-ms-error-code", equalTo("LeaseIdMissing")); + + given().body("update").header("x-ms-lease-id", "99999999-9999-9999-9999-999999999999") + .put("/{account}/{container}/{blob}", ACCOUNT, CONTAINER, BLOB) + .then().statusCode(412) + .header("x-ms-error-code", equalTo("LeaseIdMismatchWithBlobOperation")); + + given().body("update").header("x-ms-lease-id", id) + .put("/{account}/{container}/{blob}", ACCOUNT, CONTAINER, BLOB) + .then().statusCode(201); + } + + @Test + void writeWithLeaseIdWhenNotLeasedReturnsLeaseNotPresent() { + given().body("update").header("x-ms-lease-id", "99999999-9999-9999-9999-999999999999") + .put("/{account}/{container}/{blob}", ACCOUNT, CONTAINER, BLOB) + .then().statusCode(412) + .header("x-ms-error-code", equalTo("LeaseNotPresentWithBlobOperation")); + } + + @Test + void setMetadataAndDeleteHonorLeaseGuards() { + String id = acquire(); + + given().header("x-ms-meta-owner", "host-a") + .put("/{account}/{container}/{blob}?comp=metadata", ACCOUNT, CONTAINER, BLOB) + .then().statusCode(412).header("x-ms-error-code", equalTo("LeaseIdMissing")); + + given().header("x-ms-meta-owner", "host-a").header("x-ms-lease-id", id) + .put("/{account}/{container}/{blob}?comp=metadata", ACCOUNT, CONTAINER, BLOB) + .then().statusCode(200); + + given().delete("/{account}/{container}/{blob}", ACCOUNT, CONTAINER, BLOB) + .then().statusCode(412).header("x-ms-error-code", equalTo("LeaseIdMissing")); + + given().header("x-ms-lease-id", id) + .delete("/{account}/{container}/{blob}", ACCOUNT, CONTAINER, BLOB) + .then().statusCode(202); + + // Lease dies with the blob: recreate and a fresh acquire must succeed. + given().body("lock").put("/{account}/{container}/{blob}", ACCOUNT, CONTAINER, BLOB) + .then().statusCode(201); + leaseOp("acquire", "x-ms-lease-duration", "-1").then().statusCode(201); + } + + @Test + void readsRemainAllowedWhileLeased() { + acquire(); + given().get("/{account}/{container}/{blob}", ACCOUNT, CONTAINER, BLOB) + .then().statusCode(200).body(equalTo("lock")); + } +} diff --git a/src/test/java/io/floci/az/services/blob/BlobLeaseStateTest.java b/src/test/java/io/floci/az/services/blob/BlobLeaseStateTest.java new file mode 100644 index 00000000..83829408 --- /dev/null +++ b/src/test/java/io/floci/az/services/blob/BlobLeaseStateTest.java @@ -0,0 +1,61 @@ +package io.floci.az.services.blob; + +import org.junit.jupiter.api.Test; + +import java.time.Instant; + +import static org.junit.jupiter.api.Assertions.*; + +/** Pure state-machine tests: expiry and break timing as a function of (lease, now). */ +class BlobLeaseStateTest { + + private static final Instant T0 = Instant.parse("2026-01-01T00:00:00Z"); + private static final String ID = "11111111-2222-3333-4444-555555555555"; + + @Test + void infiniteLeaseStaysLeased() { + BlobLease lease = BlobLease.acquire(ID, -1, T0); + assertEquals(BlobLease.State.LEASED, lease.stateAt(T0.plusSeconds(999_999))); + assertTrue(lease.activeAt(T0.plusSeconds(999_999))); + } + + @Test + void fixedLeaseExpiresAfterDuration() { + BlobLease lease = BlobLease.acquire(ID, 15, T0); + assertEquals(BlobLease.State.LEASED, lease.stateAt(T0.plusSeconds(14))); + assertEquals(BlobLease.State.EXPIRED, lease.stateAt(T0.plusSeconds(15))); + assertFalse(lease.activeAt(T0.plusSeconds(15))); + } + + @Test + void renewResetsExpiry() { + BlobLease lease = BlobLease.acquire(ID, 15, T0).renewed(T0.plusSeconds(14)); + assertEquals(BlobLease.State.LEASED, lease.stateAt(T0.plusSeconds(20))); + assertEquals(BlobLease.State.EXPIRED, lease.stateAt(T0.plusSeconds(29))); + } + + @Test + void breakWithZeroPeriodIsImmediatelyBroken() { + BlobLease lease = BlobLease.acquire(ID, -1, T0).broken(0, T0); + assertEquals(BlobLease.State.BROKEN, lease.stateAt(T0)); + assertFalse(lease.activeAt(T0)); + assertEquals(0, lease.remainingBreakSeconds(T0)); + } + + @Test + void breakWithPeriodIsBreakingUntilElapsed() { + BlobLease lease = BlobLease.acquire(ID, -1, T0).broken(10, T0); + assertEquals(BlobLease.State.BREAKING, lease.stateAt(T0.plusSeconds(9))); + assertTrue(lease.activeAt(T0.plusSeconds(9))); + assertEquals(10, lease.remainingBreakSeconds(T0)); + assertEquals(BlobLease.State.BROKEN, lease.stateAt(T0.plusSeconds(10))); + assertFalse(lease.activeAt(T0.plusSeconds(10))); + } + + @Test + void changeKeepsTimingButSwapsId() { + BlobLease lease = BlobLease.acquire(ID, 15, T0).changed("aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"); + assertEquals("aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", lease.leaseId()); + assertEquals(BlobLease.State.EXPIRED, lease.stateAt(T0.plusSeconds(15))); + } +} From 4530ab748dfbffdde966a4073fb896187d102f7c Mon Sep 17 00:00:00 2001 From: Craig McConomy Date: Sun, 2 Aug 2026 21:50:31 -0400 Subject: [PATCH 2/6] fix(blob): close lease-guard race and validate lease headers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-ups on #183 (all three findings verified real): - Guard/mutation atomicity: replace the check-then-act validateWrite pattern with BlobLeaseService.guardedWrite(), which runs the lease guard AND the store mutation under the same monitor as the (already synchronized) lease operations. A competing acquire/break can no longer slip between the guard and the write — the interleaving the Durable partition-steal path exercises. validateWrite is now private. - change without x-ms-proposed-lease-id previously stored a lease with a null id; the next renew then NPE'd into a 500 with the lease stuck forever. renew/change/release now require x-ms-lease-id and change requires x-ms-proposed-lease-id (400 MissingRequiredHeader), and proposed ids must be GUIDs (400 InvalidHeaderValue) so a malformed id can never enter the lease store. - x-ms-lease-break-period was clamped at 0 and unbounded above, allowing a lease stuck breaking arbitrarily long. Out-of-range values (outside 0-60) now return 400 InvalidHeaderValue. Four new regression tests (watched fail first). Full suite 579 green; Java BlobLeaseClient compat and the .NET Azure.Storage.Blobs harness re-verified against the rebuilt build (still line-for-line with Azurite 3.35.0). --- .../az/services/blob/BlobLeaseService.java | 68 +++++- .../az/services/blob/BlobServiceHandler.java | 195 ++++++++---------- .../io/floci/az/services/BlobLeaseTest.java | 45 ++++ 3 files changed, 193 insertions(+), 115 deletions(-) diff --git a/src/main/java/io/floci/az/services/blob/BlobLeaseService.java b/src/main/java/io/floci/az/services/blob/BlobLeaseService.java index 03e4518e..557399e7 100644 --- a/src/main/java/io/floci/az/services/blob/BlobLeaseService.java +++ b/src/main/java/io/floci/az/services/blob/BlobLeaseService.java @@ -40,9 +40,7 @@ public synchronized Response handleLeaseOp(AzureRequest request, String blobKey, case "change" -> change(request, blobKey, lease, now); case "release" -> release(request, blobKey, lease, now); case "break" -> breakLease(request, blobKey, lease, now); - default -> new AzureErrorResponse("InvalidHeaderValue", - "The value for one of the HTTP headers is not in the correct format.") - .toXmlResponse(400); + default -> invalidHeaderValue(); }; if (response.getStatus() >= 400) { return response; @@ -62,11 +60,12 @@ private Response acquire(AzureRequest request, String blobKey, BlobLease lease, duration = 0; } if (duration != -1 && (duration < 15 || duration > 60)) { - return new AzureErrorResponse("InvalidHeaderValue", - "The value for one of the HTTP headers is not in the correct format.") - .toXmlResponse(400); + return invalidHeaderValue(); } String proposed = header(request, "x-ms-proposed-lease-id"); + if (proposed != null && !isGuid(proposed)) { + return invalidHeaderValue(); + } if (lease != null && lease.activeAt(now)) { boolean reacquireSameId = lease.stateAt(now) == BlobLease.State.LEASED @@ -86,6 +85,9 @@ private Response acquire(AzureRequest request, String blobKey, BlobLease lease, private Response renew(AzureRequest request, String blobKey, BlobLease lease, Instant now) { String leaseId = header(request, "x-ms-lease-id"); + if (leaseId == null || leaseId.isBlank()) { + return missingRequiredHeader(); + } if (lease == null) { return leaseNotPresent(); } @@ -107,6 +109,12 @@ private Response renew(AzureRequest request, String blobKey, BlobLease lease, In private Response change(AzureRequest request, String blobKey, BlobLease lease, Instant now) { String leaseId = header(request, "x-ms-lease-id"); String proposed = header(request, "x-ms-proposed-lease-id"); + if (leaseId == null || leaseId.isBlank() || proposed == null || proposed.isBlank()) { + return missingRequiredHeader(); + } + if (!isGuid(proposed)) { + return invalidHeaderValue(); + } if (lease == null || !lease.activeAt(now)) { return leaseNotPresent(); } @@ -122,6 +130,9 @@ private Response change(AzureRequest request, String blobKey, BlobLease lease, I private Response release(AzureRequest request, String blobKey, BlobLease lease, Instant now) { String leaseId = header(request, "x-ms-lease-id"); + if (leaseId == null || leaseId.isBlank()) { + return missingRequiredHeader(); + } if (lease == null) { return leaseNotPresent(); } @@ -142,11 +153,12 @@ private Response breakLease(AzureRequest request, String blobKey, BlobLease leas String breakPeriodHeader = header(request, "x-ms-lease-break-period"); if (breakPeriodHeader != null) { try { - breakPeriod = Math.max(0, Integer.parseInt(breakPeriodHeader)); + breakPeriod = Integer.parseInt(breakPeriodHeader); } catch (NumberFormatException e) { - return new AzureErrorResponse("InvalidHeaderValue", - "The value for one of the HTTP headers is not in the correct format.") - .toXmlResponse(400); + return invalidHeaderValue(); + } + if (breakPeriod < 0 || breakPeriod > 60) { + return invalidHeaderValue(); } } else { // Default: infinite leases break immediately, fixed leases run out their term. @@ -166,11 +178,45 @@ private static Response leaseNotPresent() { "There is currently no lease on the blob.").toXmlResponse(409); } + private static Response missingRequiredHeader() { + return new AzureErrorResponse("MissingRequiredHeader", + "An HTTP header that's mandatory for this request is not specified.") + .toXmlResponse(400); + } + + private static Response invalidHeaderValue() { + return new AzureErrorResponse("InvalidHeaderValue", + "The value for one of the HTTP headers is not in the correct format.") + .toXmlResponse(400); + } + + private static boolean isGuid(String value) { + try { + UUID.fromString(value); + return true; + } catch (IllegalArgumentException e) { + return false; + } + } + + /** + * Run a blob mutation under the lease guard, atomically with respect to + * lease transitions: this holds the same monitor as {@link #handleLeaseOp}, + * so a competing acquire/break cannot slip between the guard check and the + * store mutation. Returns the guard's 412 instead when the lease forbids + * the write. + */ + public synchronized Response guardedWrite(AzureRequest request, String blobKey, + java.util.function.Supplier operation) { + Response failure = validateWrite(request, blobKey); + return failure != null ? failure : operation.get(); + } + /** * Lease guard for write/delete operations on a blob. Returns null when the * operation may proceed, otherwise the 412 the Blob service contract requires. */ - public Response validateWrite(AzureRequest request, String blobKey) { + private Response validateWrite(AzureRequest request, String blobKey) { String requestLeaseId = header(request, "x-ms-lease-id"); BlobLease lease = leases.get(blobKey); Instant now = Instant.now(); diff --git a/src/main/java/io/floci/az/services/blob/BlobServiceHandler.java b/src/main/java/io/floci/az/services/blob/BlobServiceHandler.java index bf35b878..cd912f68 100644 --- a/src/main/java/io/floci/az/services/blob/BlobServiceHandler.java +++ b/src/main/java/io/floci/az/services/blob/BlobServiceHandler.java @@ -366,35 +366,32 @@ private Response putBlob(AzureRequest request, String containerName, String blob if (conditionFailure != null) { return conditionFailure; } - Response leaseFailure = leaseService.validateWrite(request, - objKey(request.accountName(), containerName, blobName)); - if (leaseFailure != null) { - return leaseFailure; - } - byte[] data = request.bodyStream().readAllBytes(); - Map metadata = new HashMap<>(); - String blobType = request.headers().getHeaderString("x-ms-blob-type"); - metadata.put("BlobType", blobType != null ? blobType : "BlockBlob"); - addBlobHttpProperties(request, metadata); - String dataLakeResourceType = request.queryParams().get("resource"); - if ("file".equals(dataLakeResourceType) || "directory".equals(dataLakeResourceType)) { - metadata.put("DataLakeResourceType", dataLakeResourceType); - } - metadata.put("Name", blobName); - metadata.put(CREATION_TIME_KEY, createdOn(existing).toString()); - metadata.putAll(readUserMetadata(request)); - - String etag = UUID.randomUUID().toString(); - store.put(objKey(request.accountName(), containerName, blobName), - new StoredObject(blobName, data, metadata, Instant.now(), etag)); - - return Response.status(Response.Status.CREATED) - .header("Last-Modified", RFC1123_DATE_TIME.format(Instant.now())) - .header("ETag", etag) - .header("x-ms-request-server-encrypted", "true") - .header("Content-Length", 0) - .build(); + return leaseService.guardedWrite(request, + objKey(request.accountName(), containerName, blobName), () -> { + Map metadata = new HashMap<>(); + String blobType = request.headers().getHeaderString("x-ms-blob-type"); + metadata.put("BlobType", blobType != null ? blobType : "BlockBlob"); + addBlobHttpProperties(request, metadata); + String dataLakeResourceType = request.queryParams().get("resource"); + if ("file".equals(dataLakeResourceType) || "directory".equals(dataLakeResourceType)) { + metadata.put("DataLakeResourceType", dataLakeResourceType); + } + metadata.put("Name", blobName); + metadata.put(CREATION_TIME_KEY, createdOn(existing).toString()); + metadata.putAll(readUserMetadata(request)); + + String etag = UUID.randomUUID().toString(); + store.put(objKey(request.accountName(), containerName, blobName), + new StoredObject(blobName, data, metadata, Instant.now(), etag)); + + return Response.status(Response.Status.CREATED) + .header("Last-Modified", RFC1123_DATE_TIME.format(Instant.now())) + .header("ETag", etag) + .header("x-ms-request-server-encrypted", "true") + .header("Content-Length", 0) + .build(); + }); } catch (IOException e) { return Response.serverError().build(); } @@ -499,14 +496,12 @@ private Response deleteBlob(AzureRequest request, String containerName, String b if (conditionFailure != null) { return conditionFailure; } - Response leaseFailure = leaseService.validateWrite(request, - objKey(request.accountName(), containerName, blobName)); - if (leaseFailure != null) { - return leaseFailure; - } - store.delete(objKey(request.accountName(), containerName, blobName)); - leaseService.onBlobDeleted(objKey(request.accountName(), containerName, blobName)); - return Response.status(Response.Status.ACCEPTED).build(); + return leaseService.guardedWrite(request, + objKey(request.accountName(), containerName, blobName), () -> { + store.delete(objKey(request.accountName(), containerName, blobName)); + leaseService.onBlobDeleted(objKey(request.accountName(), containerName, blobName)); + return Response.status(Response.Status.ACCEPTED).build(); + }); } private Response getBlobMetadata(AzureRequest request, String containerName, String blobName) { @@ -549,29 +544,26 @@ private Response setBlobMetadata(AzureRequest request, String containerName, Str return conditionFailure; } - Response leaseFailure = leaseService.validateWrite(request, - objKey(request.accountName(), containerName, blobName)); - if (leaseFailure != null) { - return leaseFailure; - } - - StoredObject so = object.get(); - Map metadata = new HashMap<>(); - so.metadata().forEach((key, value) -> { - if (!key.startsWith(USER_METADATA_PREFIX)) { - metadata.put(key, value); - } - }); - metadata.putAll(readUserMetadata(request)); + return leaseService.guardedWrite(request, + objKey(request.accountName(), containerName, blobName), () -> { + StoredObject so = object.get(); + Map metadata = new HashMap<>(); + so.metadata().forEach((key, value) -> { + if (!key.startsWith(USER_METADATA_PREFIX)) { + metadata.put(key, value); + } + }); + metadata.putAll(readUserMetadata(request)); - String etag = UUID.randomUUID().toString(); - store.put(objKey(request.accountName(), containerName, blobName), - new StoredObject(so.key(), so.data(), metadata, Instant.now(), etag)); + String etag = UUID.randomUUID().toString(); + store.put(objKey(request.accountName(), containerName, blobName), + new StoredObject(so.key(), so.data(), metadata, Instant.now(), etag)); - return Response.ok() - .header("Last-Modified", RFC1123_DATE_TIME.format(Instant.now())) - .header("ETag", etag) - .build(); + return Response.ok() + .header("Last-Modified", RFC1123_DATE_TIME.format(Instant.now())) + .header("ETag", etag) + .build(); + }); } private Response listBlobs(AzureRequest request, String containerName) { @@ -723,19 +715,17 @@ private Response putBlock(AzureRequest request, String containerName, String blo "Value for one of the query parameters specified in the request URI is invalid.") .toXmlResponse(400); } - Response leaseFailure = leaseService.validateWrite(request, - objKey(request.accountName(), containerName, blobName)); - if (leaseFailure != null) { - return leaseFailure; - } byte[] data = request.bodyStream().readAllBytes(); - store.put(blockStagingKey(request.accountName(), containerName, blobName, blockId), - new StoredObject(blockId, data, Map.of("BlockId", blockId), Instant.now(), - UUID.randomUUID().toString())); - return Response.status(Response.Status.CREATED) - .header("x-ms-request-server-encrypted", "true") - .header("Content-Length", 0) - .build(); + return leaseService.guardedWrite(request, + objKey(request.accountName(), containerName, blobName), () -> { + store.put(blockStagingKey(request.accountName(), containerName, blobName, blockId), + new StoredObject(blockId, data, Map.of("BlockId", blockId), Instant.now(), + UUID.randomUUID().toString())); + return Response.status(Response.Status.CREATED) + .header("x-ms-request-server-encrypted", "true") + .header("Content-Length", 0) + .build(); + }); } catch (IOException e) { LOGGER.errorf(e, "putBlock I/O error: container=%s blob=%s", containerName, blobName); return Response.serverError().build(); @@ -758,12 +748,6 @@ private Response putBlockList(AzureRequest request, String containerName, String .toXmlResponse(Response.Status.NOT_FOUND.getStatusCode()); } - Response leaseFailure = leaseService.validateWrite(request, - objKey(request.accountName(), containerName, blobName)); - if (leaseFailure != null) { - return leaseFailure; - } - List blockIds = parseBlockList(request.bodyStream().readAllBytes()); // Resolve every block ID → staged data @@ -792,35 +776,38 @@ private Response putBlockList(AzureRequest request, String containerName, String offset += chunk.length; } - // Build blob metadata - Map metadata = new HashMap<>(); - String blobType = request.headers().getHeaderString("x-ms-blob-type"); - metadata.put("BlobType", blobType != null ? blobType : "BlockBlob"); - addBlobHttpProperties(request, metadata); - metadata.put("Name", blobName); - metadata.put(CREATION_TIME_KEY, createdOn( - store.get(objKey(request.accountName(), containerName, blobName))).toString()); - // Persist committed block list for future GetBlockList calls - metadata.put("CommittedBlocks", String.join("|", committedMeta)); - metadata.putAll(readUserMetadata(request)); - - String etag = UUID.randomUUID().toString(); - store.put(objKey(request.accountName(), containerName, blobName), - new StoredObject(blobName, assembled, metadata, Instant.now(), etag)); - - // Clean up all staged blocks for this blob - String stagePrefix = blockStagingPrefix(request.accountName(), containerName, blobName); - store.keys().stream() - .filter(k -> k.startsWith(stagePrefix)) - .toList() - .forEach(store::delete); - - return Response.status(Response.Status.CREATED) - .header("Last-Modified", RFC1123_DATE_TIME.format(Instant.now())) - .header("ETag", etag) - .header("x-ms-request-server-encrypted", "true") - .header("Content-Length", 0) - .build(); + return leaseService.guardedWrite(request, + objKey(request.accountName(), containerName, blobName), () -> { + // Build blob metadata + Map metadata = new HashMap<>(); + String blobType = request.headers().getHeaderString("x-ms-blob-type"); + metadata.put("BlobType", blobType != null ? blobType : "BlockBlob"); + addBlobHttpProperties(request, metadata); + metadata.put("Name", blobName); + metadata.put(CREATION_TIME_KEY, createdOn( + store.get(objKey(request.accountName(), containerName, blobName))).toString()); + // Persist committed block list for future GetBlockList calls + metadata.put("CommittedBlocks", String.join("|", committedMeta)); + metadata.putAll(readUserMetadata(request)); + + String etag = UUID.randomUUID().toString(); + store.put(objKey(request.accountName(), containerName, blobName), + new StoredObject(blobName, assembled, metadata, Instant.now(), etag)); + + // Clean up all staged blocks for this blob + String stagePrefix = blockStagingPrefix(request.accountName(), containerName, blobName); + store.keys().stream() + .filter(k -> k.startsWith(stagePrefix)) + .toList() + .forEach(store::delete); + + return Response.status(Response.Status.CREATED) + .header("Last-Modified", RFC1123_DATE_TIME.format(Instant.now())) + .header("ETag", etag) + .header("x-ms-request-server-encrypted", "true") + .header("Content-Length", 0) + .build(); + }); } catch (IOException e) { LOGGER.errorf(e, "putBlockList I/O error: container=%s blob=%s", containerName, blobName); return Response.serverError().build(); diff --git a/src/test/java/io/floci/az/services/BlobLeaseTest.java b/src/test/java/io/floci/az/services/BlobLeaseTest.java index 039d25c5..f36d0efd 100644 --- a/src/test/java/io/floci/az/services/BlobLeaseTest.java +++ b/src/test/java/io/floci/az/services/BlobLeaseTest.java @@ -216,4 +216,49 @@ void readsRemainAllowedWhileLeased() { given().get("/{account}/{container}/{blob}", ACCOUNT, CONTAINER, BLOB) .then().statusCode(200).body(equalTo("lock")); } + + // ── Malformed lease headers ────────────────────────────────────────────── + + @Test + void changeWithoutProposedIdIsRejectedAndLeaseSurvives() { + String id = acquire(); + leaseOp("change", "x-ms-lease-id", id) + .then().statusCode(400).header("x-ms-error-code", equalTo("MissingRequiredHeader")); + + // The lease must not be orphaned by the rejected change. + leaseOp("renew", "x-ms-lease-id", id).then().statusCode(200); + } + + @Test + void leaseOpsWithoutLeaseIdHeaderAreRejected() { + acquire(); + leaseOp("renew") + .then().statusCode(400).header("x-ms-error-code", equalTo("MissingRequiredHeader")); + leaseOp("release") + .then().statusCode(400).header("x-ms-error-code", equalTo("MissingRequiredHeader")); + leaseOp("change", "x-ms-proposed-lease-id", "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee") + .then().statusCode(400).header("x-ms-error-code", equalTo("MissingRequiredHeader")); + } + + @Test + void breakPeriodOutsideRangeIsRejected() { + String id = acquire(); + leaseOp("break", "x-ms-lease-break-period", "99999") + .then().statusCode(400).header("x-ms-error-code", equalTo("InvalidHeaderValue")); + leaseOp("break", "x-ms-lease-break-period", "-1") + .then().statusCode(400).header("x-ms-error-code", equalTo("InvalidHeaderValue")); + + // Still an active lease after the rejected breaks. + leaseOp("renew", "x-ms-lease-id", id).then().statusCode(200); + } + + @Test + void malformedProposedLeaseIdIsRejected() { + leaseOp("acquire", "x-ms-lease-duration", "-1", "x-ms-proposed-lease-id", "not-a-guid") + .then().statusCode(400).header("x-ms-error-code", equalTo("InvalidHeaderValue")); + + String id = acquire(); + leaseOp("change", "x-ms-lease-id", id, "x-ms-proposed-lease-id", "not-a-guid") + .then().statusCode(400).header("x-ms-error-code", equalTo("InvalidHeaderValue")); + } } From fdf7d65d48d0bd876c9d8256f24f8d0ed2247af6 Mon Sep 17 00:00:00 2001 From: Craig McConomy Date: Sun, 2 Aug 2026 21:59:59 -0400 Subject: [PATCH 3/6] fix(blob): linearize container deletion and lease ops on one monitor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-up on #183: deleteContainer swept blob and lease state outside the monitor that lease operations and guarded writes hold, so a concurrent acquire could install a lease for a just-deleted blob (a recreated blob would inherit it) and a guarded write could re-put blob data into a deleted container after the sweep. Generalize guardedWrite into BlobLeaseService.exclusively(): every mutation of blob or lease state now runs its precondition checks (container/blob existence, conditional headers, lease guard) AND its store mutation inside the single lease monitor — putBlob, setBlobMetadata, deleteBlob, putBlock, putBlockList, leaseBlob's existence check, and the deleteContainer sweep. This also restores Azure's natural check order (404 before the lease 412s, which guardedWrite inverted). The remaining lease-map maintenance hooks (onBlobDeleted/onContainerDeleted/clear) are synchronized on the same monitor. New sequential regression test pins the sweep invariant: lease, delete container, recreate container+blob, write and re-acquire must succeed. Full suite 580 green; Java BlobLeaseClient compat and the .NET Azure.Storage.Blobs harness re-verified (line-for-line with Azurite). --- .../az/services/blob/BlobLeaseService.java | 29 +-- .../az/services/blob/BlobServiceHandler.java | 204 ++++++++++-------- .../io/floci/az/services/BlobLeaseTest.java | 14 ++ 3 files changed, 146 insertions(+), 101 deletions(-) diff --git a/src/main/java/io/floci/az/services/blob/BlobLeaseService.java b/src/main/java/io/floci/az/services/blob/BlobLeaseService.java index 557399e7..ff56ae0a 100644 --- a/src/main/java/io/floci/az/services/blob/BlobLeaseService.java +++ b/src/main/java/io/floci/az/services/blob/BlobLeaseService.java @@ -200,23 +200,24 @@ private static boolean isGuid(String value) { } /** - * Run a blob mutation under the lease guard, atomically with respect to - * lease transitions: this holds the same monitor as {@link #handleLeaseOp}, - * so a competing acquire/break cannot slip between the guard check and the - * store mutation. Returns the guard's 412 instead when the lease forbids - * the write. + * Run a blob operation atomically with respect to lease transitions: this + * holds the same monitor as {@link #handleLeaseOp}, so a competing + * acquire/break/release cannot interleave with the supplied operation. + * Every mutation of blob or lease state — including its precondition + * checks (existence, conditional headers, {@link #validateWrite}) — must + * run inside this to stay linearized with lease operations and with + * container deletion sweeps. */ - public synchronized Response guardedWrite(AzureRequest request, String blobKey, - java.util.function.Supplier operation) { - Response failure = validateWrite(request, blobKey); - return failure != null ? failure : operation.get(); + public synchronized Response exclusively(java.util.function.Supplier operation) { + return operation.get(); } /** * Lease guard for write/delete operations on a blob. Returns null when the - * operation may proceed, otherwise the 412 the Blob service contract requires. + * operation may proceed, otherwise the 412 the Blob service contract + * requires. Call only inside {@link #exclusively}. */ - private Response validateWrite(AzureRequest request, String blobKey) { + Response validateWrite(AzureRequest request, String blobKey) { String requestLeaseId = header(request, "x-ms-lease-id"); BlobLease lease = leases.get(blobKey); Instant now = Instant.now(); @@ -259,15 +260,15 @@ public void addLeaseHeaders(Response.ResponseBuilder rb, String blobKey) { } } - public void onBlobDeleted(String blobKey) { + public synchronized void onBlobDeleted(String blobKey) { leases.remove(blobKey); } - public void onContainerDeleted(String blobKeyPrefix) { + public synchronized void onContainerDeleted(String blobKeyPrefix) { leases.keySet().removeIf(k -> k.startsWith(blobKeyPrefix)); } - public void clear() { + public synchronized void clear() { leases.clear(); } diff --git a/src/main/java/io/floci/az/services/blob/BlobServiceHandler.java b/src/main/java/io/floci/az/services/blob/BlobServiceHandler.java index cd912f68..61714488 100644 --- a/src/main/java/io/floci/az/services/blob/BlobServiceHandler.java +++ b/src/main/java/io/floci/az/services/blob/BlobServiceHandler.java @@ -218,13 +218,17 @@ private Response notImplemented() { /** PUT /{container}/{blob}?comp=lease — Lease Blob (acquire/renew/change/release/break). */ private Response leaseBlob(AzureRequest request, String containerName, String blobName) { - Optional object = store.get(objKey(request.accountName(), containerName, blobName)); - if (object.isEmpty()) { - return new AzureErrorResponse("BlobNotFound", "The specified blob does not exist.") - .toXmlResponse(Response.Status.NOT_FOUND.getStatusCode()); - } - return leaseService.handleLeaseOp(request, objKey(request.accountName(), containerName, blobName), - object.get().etag(), RFC1123_DATE_TIME.format(object.get().lastModified())); + // The existence check must share the lease monitor, or an acquire can + // install a lease for a blob a concurrent delete just removed. + return leaseService.exclusively(() -> { + Optional object = store.get(objKey(request.accountName(), containerName, blobName)); + if (object.isEmpty()) { + return new AzureErrorResponse("BlobNotFound", "The specified blob does not exist.") + .toXmlResponse(Response.Status.NOT_FOUND.getStatusCode()); + } + return leaseService.handleLeaseOp(request, objKey(request.accountName(), containerName, blobName), + object.get().etag(), RFC1123_DATE_TIME.format(object.get().lastModified())); + }); } /** @@ -312,15 +316,19 @@ private Response deleteContainer(AzureRequest request, String containerName) { if (authFailure != null) { return authFailure; } - store.delete(nsKey(request.accountName(), containerName)); - String objPrefix = request.accountName() + "/" + containerName + "/"; - String blkPrefix = BLK_PREFIX + objPrefix; - store.keys().stream() - .filter(k -> k.startsWith(objPrefix) || k.startsWith(blkPrefix)) - .toList() - .forEach(store::delete); - leaseService.onContainerDeleted(objPrefix); - return Response.status(Response.Status.ACCEPTED).build(); + // The sweep runs under the lease monitor so no lease op or guarded + // write can interleave and resurrect blob or lease state mid-deletion. + return leaseService.exclusively(() -> { + store.delete(nsKey(request.accountName(), containerName)); + String objPrefix = request.accountName() + "/" + containerName + "/"; + String blkPrefix = BLK_PREFIX + objPrefix; + store.keys().stream() + .filter(k -> k.startsWith(objPrefix) || k.startsWith(blkPrefix)) + .toList() + .forEach(store::delete); + leaseService.onContainerDeleted(objPrefix); + return Response.status(Response.Status.ACCEPTED).build(); + }); } private Response listContainers(AzureRequest request) { @@ -350,25 +358,30 @@ private Response listContainers(AzureRequest request) { private Response putBlob(AzureRequest request, String containerName, String blobName) { try { - Optional existing = store.get(objKey(request.accountName(), containerName, blobName)); - Response authFailure = existing.isPresent() + Response authFailure = store.get(objKey(request.accountName(), containerName, blobName)).isPresent() ? authorizeWrite(request, containerName, blobName) : authorizeCreate(request, containerName, blobName); if (authFailure != null) { return authFailure; } - if (store.get(nsKey(request.accountName(), containerName)).isEmpty()) { - return new AzureErrorResponse("ContainerNotFound", "The specified container does not exist.") - .toXmlResponse(Response.Status.NOT_FOUND.getStatusCode()); - } - - Response conditionFailure = validateBlobConditions(request, existing); - if (conditionFailure != null) { - return conditionFailure; - } byte[] data = request.bodyStream().readAllBytes(); - return leaseService.guardedWrite(request, - objKey(request.accountName(), containerName, blobName), () -> { + return leaseService.exclusively(() -> { + if (store.get(nsKey(request.accountName(), containerName)).isEmpty()) { + return new AzureErrorResponse("ContainerNotFound", "The specified container does not exist.") + .toXmlResponse(Response.Status.NOT_FOUND.getStatusCode()); + } + + Optional existing = store.get(objKey(request.accountName(), containerName, blobName)); + Response conditionFailure = validateBlobConditions(request, existing); + if (conditionFailure != null) { + return conditionFailure; + } + Response leaseFailure = leaseService.validateWrite(request, + objKey(request.accountName(), containerName, blobName)); + if (leaseFailure != null) { + return leaseFailure; + } + Map metadata = new HashMap<>(); String blobType = request.headers().getHeaderString("x-ms-blob-type"); metadata.put("BlobType", blobType != null ? blobType : "BlockBlob"); @@ -487,17 +500,21 @@ private Response deleteBlob(AzureRequest request, String containerName, String b if (authFailure != null) { return authFailure; } - Optional object = store.get(objKey(request.accountName(), containerName, blobName)); - if (object.isEmpty()) { - return new AzureErrorResponse("BlobNotFound", "The specified blob does not exist.") - .toXmlResponse(Response.Status.NOT_FOUND.getStatusCode()); - } - Response conditionFailure = validateBlobConditions(request, object); - if (conditionFailure != null) { - return conditionFailure; - } - return leaseService.guardedWrite(request, - objKey(request.accountName(), containerName, blobName), () -> { + return leaseService.exclusively(() -> { + Optional object = store.get(objKey(request.accountName(), containerName, blobName)); + if (object.isEmpty()) { + return new AzureErrorResponse("BlobNotFound", "The specified blob does not exist.") + .toXmlResponse(Response.Status.NOT_FOUND.getStatusCode()); + } + Response conditionFailure = validateBlobConditions(request, object); + if (conditionFailure != null) { + return conditionFailure; + } + Response leaseFailure = leaseService.validateWrite(request, + objKey(request.accountName(), containerName, blobName)); + if (leaseFailure != null) { + return leaseFailure; + } store.delete(objKey(request.accountName(), containerName, blobName)); leaseService.onBlobDeleted(objKey(request.accountName(), containerName, blobName)); return Response.status(Response.Status.ACCEPTED).build(); @@ -533,19 +550,23 @@ private Response setBlobMetadata(AzureRequest request, String containerName, Str if (authFailure != null) { return authFailure; } - Optional object = store.get(objKey(request.accountName(), containerName, blobName)); - if (object.isEmpty()) { - return new AzureErrorResponse("BlobNotFound", "The specified blob does not exist.") - .toXmlResponse(Response.Status.NOT_FOUND.getStatusCode()); - } + return leaseService.exclusively(() -> { + Optional object = store.get(objKey(request.accountName(), containerName, blobName)); + if (object.isEmpty()) { + return new AzureErrorResponse("BlobNotFound", "The specified blob does not exist.") + .toXmlResponse(Response.Status.NOT_FOUND.getStatusCode()); + } - Response conditionFailure = validateBlobConditions(request, object); - if (conditionFailure != null) { - return conditionFailure; - } + Response conditionFailure = validateBlobConditions(request, object); + if (conditionFailure != null) { + return conditionFailure; + } + Response leaseFailure = leaseService.validateWrite(request, + objKey(request.accountName(), containerName, blobName)); + if (leaseFailure != null) { + return leaseFailure; + } - return leaseService.guardedWrite(request, - objKey(request.accountName(), containerName, blobName), () -> { StoredObject so = object.get(); Map metadata = new HashMap<>(); so.metadata().forEach((key, value) -> { @@ -705,10 +726,6 @@ private Response putBlock(AzureRequest request, String containerName, String blo if (authFailure != null) { return authFailure; } - if (store.get(nsKey(request.accountName(), containerName)).isEmpty()) { - return new AzureErrorResponse("ContainerNotFound", "The specified container does not exist.") - .toXmlResponse(Response.Status.NOT_FOUND.getStatusCode()); - } String blockId = request.queryParams().get("blockid"); if (blockId == null || blockId.isBlank()) { return new AzureErrorResponse("InvalidQueryParameterValue", @@ -716,8 +733,16 @@ private Response putBlock(AzureRequest request, String containerName, String blo .toXmlResponse(400); } byte[] data = request.bodyStream().readAllBytes(); - return leaseService.guardedWrite(request, - objKey(request.accountName(), containerName, blobName), () -> { + return leaseService.exclusively(() -> { + if (store.get(nsKey(request.accountName(), containerName)).isEmpty()) { + return new AzureErrorResponse("ContainerNotFound", "The specified container does not exist.") + .toXmlResponse(Response.Status.NOT_FOUND.getStatusCode()); + } + Response leaseFailure = leaseService.validateWrite(request, + objKey(request.accountName(), containerName, blobName)); + if (leaseFailure != null) { + return leaseFailure; + } store.put(blockStagingKey(request.accountName(), containerName, blobName, blockId), new StoredObject(blockId, data, Map.of("BlockId", blockId), Instant.now(), UUID.randomUUID().toString())); @@ -743,41 +768,46 @@ private Response putBlockList(AzureRequest request, String containerName, String if (authFailure != null) { return authFailure; } - if (store.get(nsKey(request.accountName(), containerName)).isEmpty()) { - return new AzureErrorResponse("ContainerNotFound", "The specified container does not exist.") - .toXmlResponse(Response.Status.NOT_FOUND.getStatusCode()); - } - List blockIds = parseBlockList(request.bodyStream().readAllBytes()); - // Resolve every block ID → staged data - List chunks = new ArrayList<>(blockIds.size()); - List committedMeta = new ArrayList<>(blockIds.size()); // "base64id:size" - - for (String blockId : blockIds) { - Optional staged = store.get( - blockStagingKey(request.accountName(), containerName, blobName, blockId)); - if (staged.isEmpty()) { - return new AzureErrorResponse("InvalidBlockList", - "The specified block list is invalid.") - .toXmlResponse(400); + return leaseService.exclusively(() -> { + if (store.get(nsKey(request.accountName(), containerName)).isEmpty()) { + return new AzureErrorResponse("ContainerNotFound", "The specified container does not exist.") + .toXmlResponse(Response.Status.NOT_FOUND.getStatusCode()); } - byte[] blockData = staged.get().data(); - chunks.add(blockData); - committedMeta.add(blockId + ":" + blockData.length); - } - // Concatenate all block data into the final blob body - int totalSize = chunks.stream().mapToInt(c -> c.length).sum(); - byte[] assembled = new byte[totalSize]; - int offset = 0; - for (byte[] chunk : chunks) { - System.arraycopy(chunk, 0, assembled, offset, chunk.length); - offset += chunk.length; - } + // Resolve every block ID → staged data + List chunks = new ArrayList<>(blockIds.size()); + List committedMeta = new ArrayList<>(blockIds.size()); // "base64id:size" + + for (String blockId : blockIds) { + Optional staged = store.get( + blockStagingKey(request.accountName(), containerName, blobName, blockId)); + if (staged.isEmpty()) { + return new AzureErrorResponse("InvalidBlockList", + "The specified block list is invalid.") + .toXmlResponse(400); + } + byte[] blockData = staged.get().data(); + chunks.add(blockData); + committedMeta.add(blockId + ":" + blockData.length); + } + + // Concatenate all block data into the final blob body + int totalSize = chunks.stream().mapToInt(c -> c.length).sum(); + byte[] assembled = new byte[totalSize]; + int offset = 0; + for (byte[] chunk : chunks) { + System.arraycopy(chunk, 0, assembled, offset, chunk.length); + offset += chunk.length; + } + + Response leaseFailure = leaseService.validateWrite(request, + objKey(request.accountName(), containerName, blobName)); + if (leaseFailure != null) { + return leaseFailure; + } - return leaseService.guardedWrite(request, - objKey(request.accountName(), containerName, blobName), () -> { // Build blob metadata Map metadata = new HashMap<>(); String blobType = request.headers().getHeaderString("x-ms-blob-type"); diff --git a/src/test/java/io/floci/az/services/BlobLeaseTest.java b/src/test/java/io/floci/az/services/BlobLeaseTest.java index f36d0efd..58485d50 100644 --- a/src/test/java/io/floci/az/services/BlobLeaseTest.java +++ b/src/test/java/io/floci/az/services/BlobLeaseTest.java @@ -217,6 +217,20 @@ void readsRemainAllowedWhileLeased() { .then().statusCode(200).body(equalTo("lock")); } + @Test + void containerDeleteSweepsLeaseState() { + acquire(); + given().delete("/{account}/{container}?restype=container", ACCOUNT, CONTAINER) + .then().statusCode(202); + + // Recreate container and blob: a stale lease would make this PUT 412. + given().put("/{account}/{container}?restype=container", ACCOUNT, CONTAINER) + .then().statusCode(201); + given().body("lock").put("/{account}/{container}/{blob}", ACCOUNT, CONTAINER, BLOB) + .then().statusCode(201); + leaseOp("acquire", "x-ms-lease-duration", "-1").then().statusCode(201); + } + // ── Malformed lease headers ────────────────────────────────────────────── @Test From 74189040b311e7e21b05cf59c95172e6e9ba0bdc Mon Sep 17 00:00:00 2001 From: Craig McConomy Date: Tue, 18 Aug 2026 14:34:41 -0400 Subject: [PATCH 4/6] fix(blob): cap break period at a fixed lease's natural expiry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Break Lease applies x-ms-lease-break-period only when it is shorter than the lease's remaining time, but broken() stamped breakAt unconditionally at now + period. A fixed lease with less time remaining than the requested period therefore sat in BREAKING — rejecting reacquisition — past the moment it should have expired, since stateAt() checks breakAt before expiresAt. Cap breakAt at expiresAt in broken(). The no-header default path already derived the period from the remaining time; the explicit-header path now honors the same bound. State-machine tests pin both directions: a 60s break on a lease with 5s left completes at natural expiry, and a period shorter than the remaining time is used as given. --- .../io/floci/az/services/blob/BlobLease.java | 8 +++++++- .../az/services/blob/BlobLeaseStateTest.java | 19 +++++++++++++++++++ 2 files changed, 26 insertions(+), 1 deletion(-) diff --git a/src/main/java/io/floci/az/services/blob/BlobLease.java b/src/main/java/io/floci/az/services/blob/BlobLease.java index d950859a..3f8fc5ac 100644 --- a/src/main/java/io/floci/az/services/blob/BlobLease.java +++ b/src/main/java/io/floci/az/services/blob/BlobLease.java @@ -31,7 +31,13 @@ public BlobLease changed(String newLeaseId) { } public BlobLease broken(int breakPeriodSeconds, Instant now) { - return new BlobLease(leaseId, durationSeconds, expiresAt, now.plusSeconds(breakPeriodSeconds)); + Instant breakAt = now.plusSeconds(breakPeriodSeconds); + // The break period is used only if it is shorter than the lease's + // remaining time; a fixed lease never outlives its natural expiry. + if (expiresAt != null && expiresAt.isBefore(breakAt)) { + breakAt = expiresAt; + } + return new BlobLease(leaseId, durationSeconds, expiresAt, breakAt); } public State stateAt(Instant now) { diff --git a/src/test/java/io/floci/az/services/blob/BlobLeaseStateTest.java b/src/test/java/io/floci/az/services/blob/BlobLeaseStateTest.java index 83829408..a1445b6d 100644 --- a/src/test/java/io/floci/az/services/blob/BlobLeaseStateTest.java +++ b/src/test/java/io/floci/az/services/blob/BlobLeaseStateTest.java @@ -52,6 +52,25 @@ void breakWithPeriodIsBreakingUntilElapsed() { assertFalse(lease.activeAt(T0.plusSeconds(10))); } + @Test + void breakPeriodLongerThanRemainingTimeCapsAtNaturalExpiry() { + // 15s lease, broken at t+10 with a 60s period: the break period is used + // only if it is shorter than the remaining lease time, so the lease + // must be BROKEN at its natural expiry (t+15), not at t+70. + BlobLease lease = BlobLease.acquire(ID, 15, T0).broken(60, T0.plusSeconds(10)); + assertEquals(BlobLease.State.BREAKING, lease.stateAt(T0.plusSeconds(14))); + assertEquals(5, lease.remainingBreakSeconds(T0.plusSeconds(10))); + assertEquals(BlobLease.State.BROKEN, lease.stateAt(T0.plusSeconds(15))); + assertFalse(lease.activeAt(T0.plusSeconds(15))); + } + + @Test + void breakPeriodShorterThanRemainingTimeIsUsed() { + BlobLease lease = BlobLease.acquire(ID, 60, T0).broken(5, T0); + assertEquals(BlobLease.State.BREAKING, lease.stateAt(T0.plusSeconds(4))); + assertEquals(BlobLease.State.BROKEN, lease.stateAt(T0.plusSeconds(5))); + } + @Test void changeKeepsTimingButSwapsId() { BlobLease lease = BlobLease.acquire(ID, 15, T0).changed("aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"); From 6b19dea349d651541eb12d646f446d73fc3f7e36 Mon Sep 17 00:00:00 2001 From: Craig McConomy Date: Thu, 20 Aug 2026 11:45:18 -0400 Subject: [PATCH 5/6] fix(blob): create containers under the lease monitor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit createContainer checked and wrote the namespace sentinel outside the monitor that deleteContainer sweeps under, so a create that observed absence could re-put the sentinel after a concurrent deletion sweep — resurrecting the container after DELETE answered 202 — and two concurrent creates could both answer 201 where the contract gives one ContainerAlreadyExists. ensureContainer (ARM provisioning) had the same unsynchronized put. Run both under the lease monitor via a void exclusively(Runnable) overload, and move the admin-reset sweep (clear) under it too — the last mutation outside the monitor — so the invariant stated in exclusively()'s javadoc holds everywhere. Sequential behavior is unchanged; the interleaving is eliminated by construction and the existing container lifecycle tests pin the sequential contract. --- .../az/services/blob/BlobLeaseService.java | 5 +++ .../az/services/blob/BlobServiceHandler.java | 32 ++++++++++++------- 2 files changed, 25 insertions(+), 12 deletions(-) diff --git a/src/main/java/io/floci/az/services/blob/BlobLeaseService.java b/src/main/java/io/floci/az/services/blob/BlobLeaseService.java index ff56ae0a..b51ecab7 100644 --- a/src/main/java/io/floci/az/services/blob/BlobLeaseService.java +++ b/src/main/java/io/floci/az/services/blob/BlobLeaseService.java @@ -212,6 +212,11 @@ public synchronized Response exclusively(java.util.function.Supplier o return operation.get(); } + /** {@link #exclusively(java.util.function.Supplier)} for mutations that produce no response. */ + public synchronized void exclusively(Runnable operation) { + operation.run(); + } + /** * Lease guard for write/delete operations on a blob. Returns null when the * operation may proceed, otherwise the 412 the Blob service contract diff --git a/src/main/java/io/floci/az/services/blob/BlobServiceHandler.java b/src/main/java/io/floci/az/services/blob/BlobServiceHandler.java index 61714488..2da8197f 100644 --- a/src/main/java/io/floci/az/services/blob/BlobServiceHandler.java +++ b/src/main/java/io/floci/az/services/blob/BlobServiceHandler.java @@ -300,15 +300,21 @@ private Response createContainer(AzureRequest request, String containerName) { return authFailure; } String key = nsKey(request.accountName(), containerName); - if (store.get(key).isPresent()) { - return new AzureErrorResponse("ContainerAlreadyExists", "The specified container already exists.") - .toXmlResponse(Response.Status.CONFLICT.getStatusCode()); - } - store.put(key, NS_SENTINEL); - return Response.status(Response.Status.CREATED) - .header("Last-Modified", RFC1123_DATE_TIME.format(Instant.now())) - .header("ETag", UUID.randomUUID().toString()) - .build(); + // Check-and-create must share the lease monitor, or a create that + // observed absence can re-put the sentinel after a concurrent deletion + // sweep (resurrecting the container after DELETE answered 202), and + // two concurrent creates can both answer 201. + return leaseService.exclusively(() -> { + if (store.get(key).isPresent()) { + return new AzureErrorResponse("ContainerAlreadyExists", "The specified container already exists.") + .toXmlResponse(Response.Status.CONFLICT.getStatusCode()); + } + store.put(key, NS_SENTINEL); + return Response.status(Response.Status.CREATED) + .header("Last-Modified", RFC1123_DATE_TIME.format(Instant.now())) + .header("ETag", UUID.randomUUID().toString()) + .build(); + }); } private Response deleteContainer(AzureRequest request, String containerName) { @@ -1006,12 +1012,14 @@ private static List parseBlockList(byte[] body) { } public void clear() { - store.clear(); - leaseService.clear(); + leaseService.exclusively(() -> { + store.clear(); + leaseService.clear(); + }); } public void ensureContainer(String accountName, String containerName) { - store.put(nsKey(accountName, containerName), NS_SENTINEL); + leaseService.exclusively(() -> store.put(nsKey(accountName, containerName), NS_SENTINEL)); } private static String nsKey(String accountName, String containerName) { From 60bdf1046bb81899fc36fd77d0672930e6718667 Mon Sep 17 00:00:00 2001 From: Craig McConomy Date: Thu, 20 Aug 2026 11:55:08 -0400 Subject: [PATCH 6/6] fix(blob): classify PutBlob authorization inside the lease monitor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PutBlob chooses between create and write SAS authorization from blob existence, but that read sat outside the monitor while the mutation re-read state inside it. A create-only SAS (sp=c without w) that observed absence could keep CREATE authorization and overwrite a blob a concurrent request had just created, destroying its data. Move the classification inside exclusively(), driven by the same existing snapshot the mutation uses — one read now feeds authorization, the conditional-header checks, and the creation-time stamp. Sequential precedence is unchanged (403 before the container 404, then conditions, then the lease guard), pinned by the existing create-only SAS overwrite test; the body read stays outside the monitor so no I/O runs under the lock, and the authorize checks are pure token computation. --- .../az/services/blob/BlobServiceHandler.java | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/src/main/java/io/floci/az/services/blob/BlobServiceHandler.java b/src/main/java/io/floci/az/services/blob/BlobServiceHandler.java index 2da8197f..3bdebfae 100644 --- a/src/main/java/io/floci/az/services/blob/BlobServiceHandler.java +++ b/src/main/java/io/floci/az/services/blob/BlobServiceHandler.java @@ -364,20 +364,24 @@ private Response listContainers(AzureRequest request) { private Response putBlob(AzureRequest request, String containerName, String blobName) { try { - Response authFailure = store.get(objKey(request.accountName(), containerName, blobName)).isPresent() - ? authorizeWrite(request, containerName, blobName) - : authorizeCreate(request, containerName, blobName); - if (authFailure != null) { - return authFailure; - } byte[] data = request.bodyStream().readAllBytes(); return leaseService.exclusively(() -> { + // Create-vs-write is classified from blob existence, so the + // classification must read the same snapshot the mutation + // uses: outside the monitor, a create-only SAS that observed + // absence could overwrite a concurrently created blob. + Optional existing = store.get(objKey(request.accountName(), containerName, blobName)); + Response authFailure = existing.isPresent() + ? authorizeWrite(request, containerName, blobName) + : authorizeCreate(request, containerName, blobName); + if (authFailure != null) { + return authFailure; + } if (store.get(nsKey(request.accountName(), containerName)).isEmpty()) { return new AzureErrorResponse("ContainerNotFound", "The specified container does not exist.") .toXmlResponse(Response.Status.NOT_FOUND.getStatusCode()); } - Optional existing = store.get(objKey(request.accountName(), containerName, blobName)); Response conditionFailure = validateBlobConditions(request, existing); if (conditionFailure != null) { return conditionFailure;