feat(blob): blob lease support for the Azure Functions storage backplane

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);
+    }
 }

src/main/java/io/floci/az/services/blob/BlobLease.java

@@ -0,0 +1,66 @@
+package io.floci.az.services.blob;
+
+import java.time.Instant;
+
+/**
+ * State of one blob lease. Immutable; transitions return a new instance.
+ *
+ * <p>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) {
+        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) {
+        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();
+    }
+}