feat(storage): release unused capacity reservations

modules/jarvos-storage-janitor/README.md

@@ -56,6 +56,42 @@ contract. In short:
   inputs; reserving against a pool subtracts every already-active
   reservation held for that `poolId` across all fence generations, so two
   distinct reservations cannot double-commit the same headroom.
+  `release({ reservationId, idempotencyKey, now })` ends an active,
+  unexpired hold without any drawdown: `consumedBytes` stays zero and the
+  reservation immediately stops counting toward its pool's active headroom.
+  A same-key replay is a stable success; a different-key replay against an
+  already-released reservation, or a release attempt against a consumed,
+  expired, or missing reservation, is a typed blocked result rather than a
+  mutation. `released` is terminal -- a `reserve` idempotency replay against
+  a released reservation is rejected, never reopened, and a `consume`
+  attempt against a released reservation is rejected as `already_released`
+  with no drawdown, so no transition can ever reopen or rewrite a released
+  reservation. `reap` never reverts a released reservation back to `expired`
+  either: terminal means terminal against every transition, not just
+  `reserve` and `consume`.
+
+  `release()` and the `released` status are a **breaking addition** to the
+  reservation-persistence port and its persisted schema: the store's
+  persisted `schemaVersion` is `jarvos-storage-janitor.reservation-store.v2`.
+  Any host adapter implementing this port directly (not through
+  `createMemoryReservationStore`) must add `release()` -- see
+  `assertReservationPort` in `ports.js`. A **genuine** v1 store (one written
+  before `release` existed) contains only `active`, `consumed`, or `expired`
+  records; this package still reads it, normalizes its missing
+  `releasedAt`/`releaseIdempotencyKey` fields to `null` in memory, and
+  upgrades it to v2 in storage the next time it is mutated -- `get()` alone
+  never writes anything. A v1 store that impossibly contains a `released`
+  record (which genuine v1 code could never have written) is rejected as
+  invalid rather than silently accepted. This upgrade path is one-way: a
+  v1-only rollback boundary. Once any node has upgraded a store to v2 (by
+  mutating it, or simply by starting from an already-v2 empty state), old
+  v1-only code reading that store sees a `schemaVersion` it does not
+  recognize and fails closed with a schema-mismatch error -- it does not
+  crash, and it does not silently misinterpret a `released` record as some
+  other status. Rolling back to v1-only code against a store any v2 code has
+  touched is therefore unsafe and unsupported; a rollback requires either a
+  v1-only store that has never been touched by v2 code, or restoring a v1
+  snapshot taken before the first v2 write.
 - **Ports** (`ports.js`) define the three explicit boundaries this package
   depends on -- capacity observation, external reclaim provider, and
   reservation persistence -- as typed method-shape contracts. None receives a

modules/jarvos-storage-janitor/src/ports.js

@@ -31,11 +31,11 @@ function assertExternalReclaimPort(port) {
   assertPortShape(port, ['proposeDryRun', 'execute'], 'external reclaim port');
 }
 
-// ReservationPersistencePort: reserve/consume/reap/get, matching the
+// ReservationPersistencePort: reserve/consume/release/reap/get, matching the
 // reservation-store.js contract. A conforming implementation must satisfy
 // checkReservationStoreConformance from reservation-store.js.
 function assertReservationPort(port) {
-  assertPortShape(port, ['reserve', 'consume', 'reap', 'get'], 'reservation-persistence port');
+  assertPortShape(port, ['reserve', 'consume', 'release', 'reap', 'get'], 'reservation-persistence port');
 }
 
 module.exports = {

modules/jarvos-storage-janitor/src/reservation-store.js

@@ -2,8 +2,16 @@
 
 const { isObject, clone, isOpaqueId, isSafeNonNegativeInt, isSafePositiveInt, isValidClockValue, normalizeTime, digestOf } = require('./primitives');
 
-const RESERVATION_STORE_SCHEMA_VERSION = 'jarvos-storage-janitor.reservation-store.v1';
-const RESERVATION_STATES = Object.freeze(['active', 'consumed', 'expired']);
+// v1 shipped before `release`/`released` existed: it is the pre-release
+// contract, and a genuine v1 store can therefore never contain a `released`
+// record. v2 is a breaking addition (a new terminal status and its
+// `releasedAt`/`releaseIdempotencyKey` fields); old v1-only code must fail
+// closed on a v2 store via this schema-version mismatch rather than silently
+// misreading a status it does not know about. See README.md for the
+// documented rollback boundary this implies.
+const RESERVATION_STORE_SCHEMA_VERSION_V1 = 'jarvos-storage-janitor.reservation-store.v1';
+const RESERVATION_STORE_SCHEMA_VERSION = 'jarvos-storage-janitor.reservation-store.v2';
+const RESERVATION_STATES = Object.freeze(['active', 'consumed', 'expired', 'released']);
 const MAX_MUTATE_ATTEMPTS = 8;
 
 function emptyState() {
@@ -21,6 +29,55 @@ function validateState(state) {
   return { ok: errors.length === 0, errors };
 }
 
+// A legitimate v1 store only ever wrote `active`, `consumed`, or `expired`;
+// a `released` record under a v1 schemaVersion is impossible for genuine v1
+// data, so it is rejected as invalid rather than silently normalized.
+function validateV1State(state) {
+  const errors = [];
+  if (!isObject(state)) return { ok: false, errors: ['reservation-store state must be an object'] };
+  if (state.schemaVersion !== RESERVATION_STORE_SCHEMA_VERSION_V1) errors.push(`state.schemaVersion must be ${RESERVATION_STORE_SCHEMA_VERSION_V1}`);
+  if (!Number.isInteger(state.revision) || state.revision < 0) errors.push('state.revision must be a non-negative integer');
+  if (!Number.isInteger(state.currentFence) || state.currentFence < 0) errors.push('state.currentFence must be a non-negative integer');
+  if (!isObject(state.reservations)) errors.push('state.reservations must be an object');
+  if (!isObject(state.idempotencyIndex)) errors.push('state.idempotencyIndex must be an object');
+  if (isObject(state.reservations)) {
+    for (const record of Object.values(state.reservations)) {
+      if (isObject(record) && record.status === 'released') {
+        errors.push(`reservation ${record.reservationId} has status "released" under schemaVersion ${RESERVATION_STORE_SCHEMA_VERSION_V1}, which is impossible for a genuine v1 store`);
+      }
+    }
+  }
+  return { ok: errors.length === 0, errors };
+}
+
+// Upgrades an already-validated v1 state to the v2 shape in memory, adding
+// the `released`-status fields a v1 record never had. This does not persist
+// anything by itself: a mutation persists the upgrade via its normal save,
+// while a read-only `get` returns the upgraded shape without writing it back.
+function upgradeV1State(state) {
+  const upgraded = clone(state);
+  upgraded.schemaVersion = RESERVATION_STORE_SCHEMA_VERSION;
+  for (const record of Object.values(upgraded.reservations)) {
+    if (record.releaseIdempotencyKey === undefined) record.releaseIdempotencyKey = null;
+    if (record.releasedAt === undefined) record.releasedAt = null;
+  }
+  return upgraded;
+}
+
+// Every read path funnels through here so a legitimate v1 store loads and
+// upgrades exactly once, a v2 store loads as-is, and anything else --
+// including a v1 store impossibly marked `released` -- fails closed.
+function loadAndUpgradeState(loaded) {
+  if (isObject(loaded) && loaded.schemaVersion === RESERVATION_STORE_SCHEMA_VERSION_V1) {
+    const v1Validation = validateV1State(loaded);
+    if (!v1Validation.ok) return { ok: false, errors: v1Validation.errors };
+    return { ok: true, state: upgradeV1State(loaded) };
+  }
+  const validation = validateState(loaded);
+  if (!validation.ok) return { ok: false, errors: validation.errors };
+  return { ok: true, state: loaded };
+}
+
 // A conflict here is the store's declared atomic primitive speaking: a
 // conforming backend detects a stale compare-and-set precondition and raises
 // exactly this shape rather than silently overwriting the loser's read.
@@ -56,6 +113,7 @@ function publicReservation(record) {
     createdAt: record.createdAt,
     expiresAt: record.expiresAt,
     consumedAt: record.consumedAt,
+    releasedAt: record.releasedAt,
   };
 }
 
@@ -83,9 +141,9 @@ function createReservationStore(options = {}) {
   async function mutate(mutator) {
     for (let attempt = 0; attempt < MAX_MUTATE_ATTEMPTS; attempt += 1) {
       const loaded = await backend.load();
-      const validation = validateState(loaded);
-      if (!validation.ok) throw new Error(`invalid reservation-store state: ${validation.errors.join('; ')}`);
-      const state = clone(loaded);
+      const normalized = loadAndUpgradeState(loaded);
+      if (!normalized.ok) throw new Error(`invalid reservation-store state: ${normalized.errors.join('; ')}`);
+      const state = clone(normalized.state);
       const expectedRevision = state.revision;
       const outcome = mutator(state);
       if (outcome && outcome.__noCommit) return outcome.value;
@@ -160,6 +218,9 @@ function createReservationStore(options = {}) {
           }
           if (existing.status === 'expired') return noCommit({ ok: false, reason: 'expired' });
           if (existing.status === 'consumed') return noCommit({ ok: false, reason: 'already_consumed' });
+          // A released reservation is terminal: a reserve replay must never
+          // reopen it, regardless of matching parameters or fence.
+          if (existing.status === 'released') return noCommit({ ok: false, reason: 'already_released' });
 
           const matches = existing.amountBytes === amountBytes
             && existing.fenceGeneration === fenceGeneration
@@ -222,9 +283,11 @@ function createReservationStore(options = {}) {
           consumedBytes: 0,
           status: 'active',
           consumeIdempotencyKey: null,
+          releaseIdempotencyKey: null,
           createdAt: effectiveNow,
           expiresAt: expiresIso,
           consumedAt: null,
+          releasedAt: null,
         };
         state.reservations[id] = record;
         state.idempotencyIndex[idempotencyKey] = id;
@@ -258,6 +321,10 @@ function createReservationStore(options = {}) {
           return noCommit({ ok: true, replayed: true, reservation: publicReservation(record) });
         }
         if (record.status === 'consumed') return noCommit({ ok: false, reason: 'already_consumed' });
+        // A released reservation is terminal: a consume attempt must never
+        // reopen it and rewrite it to `consumed`, regardless of idempotency
+        // key or requested amount.
+        if (record.status === 'released') return noCommit({ ok: false, reason: 'already_released' });
         if (record.status === 'expired') return noCommit({ ok: false, reason: 'expired' });
         if (record.status === 'active' && isExpired(record, effectiveNow)) {
           // Commit the expiry transition rather than reporting `expired`
@@ -276,6 +343,49 @@ function createReservationStore(options = {}) {
     });
   }
 
+  function release({ reservationId: id, idempotencyKey, now } = {}) {
+    return guarded(async () => {
+      if (!isOpaqueId(id) || !isOpaqueId(idempotencyKey)) {
+        return { ok: false, reason: 'invalid_request', errors: ['reservationId and idempotencyKey are required and must be well-formed'] };
+      }
+      const resolvedNow = resolveNow(now, clock);
+      if (!resolvedNow.ok) return { ok: false, reason: 'invalid_request', errors: ['now must be a valid UTC ISO-8601 timestamp'] };
+      const effectiveNow = resolvedNow.value;
+
+      return mutate((state) => {
+        const record = state.reservations[id];
+        if (!record) return noCommit({ ok: false, reason: 'not_found' });
+
+        // A repeat release with the same idempotency key replays the
+        // original result; a released reservation is terminal, so a
+        // same-key replay is the only way a second release call can
+        // succeed. A different key against an already-released
+        // reservation is a typed conflict, never a silent reuse.
+        if (record.status === 'released' && record.releaseIdempotencyKey === idempotencyKey) {
+          return noCommit({ ok: true, replayed: true, reservation: publicReservation(record) });
+        }
+        if (record.status === 'released') return noCommit({ ok: false, reason: 'already_released' });
+        if (record.status === 'consumed') return noCommit({ ok: false, reason: 'already_consumed' });
+        if (record.status === 'expired') return noCommit({ ok: false, reason: 'expired' });
+        if (record.status === 'active' && isExpired(record, effectiveNow)) {
+          // Commit the expiry transition rather than reporting `expired`
+          // for a mutation that was never actually persisted.
+          record.status = 'expired';
+          return { ok: false, reason: 'expired' };
+        }
+
+        // Freeing an active, unexpired reservation drops its status out of
+        // `active`, so it stops counting toward reserve()'s aggregate
+        // active-headroom sum for its pool immediately, with no
+        // consumedBytes drawdown.
+        record.status = 'released';
+        record.releaseIdempotencyKey = idempotencyKey;
+        record.releasedAt = effectiveNow;
+        return { ok: true, replayed: false, reservation: publicReservation(record) };
+      });
+    });
+  }
+
   function reap({ now } = {}) {
     return guarded(async () => {
       const resolvedNow = resolveNow(now, clock);
@@ -299,16 +409,16 @@ function createReservationStore(options = {}) {
   function get(id) {
     return guarded(async () => {
       if (!isOpaqueId(id)) return { ok: false, reason: 'invalid_request', errors: ['reservationId must be an opaque identifier'] };
-      const state = await backend.load();
-      const validation = validateState(state);
-      if (!validation.ok) throw new Error(`invalid reservation-store state: ${validation.errors.join('; ')}`);
-      const record = state.reservations[id];
+      const loaded = await backend.load();
+      const normalized = loadAndUpgradeState(loaded);
+      if (!normalized.ok) throw new Error(`invalid reservation-store state: ${normalized.errors.join('; ')}`);
+      const record = normalized.state.reservations[id];
       if (!record) return { ok: false, reason: 'not_found' };
       return { ok: true, reservation: publicReservation(record) };
     });
   }
 
-  return { reserve, consume, reap, get };
+  return { reserve, consume, release, reap, get };
 }
 
 function createMemoryReservationBackend() {
@@ -337,7 +447,11 @@ function createMemoryReservationStore(options = {}) {
 // equivalent serialization) this store depends on for no-double-spend: a
 // second save() against a precondition its own first save() already
 // invalidated must be rejected with a reservationConflict error, not
-// silently accepted as a last-writer-wins overwrite.
+// silently accepted as a last-writer-wins overwrite. This is a mechanical
+// property of a *fresh* backend, not a business-semantics check, so it
+// validates against the current schema directly rather than through
+// loadAndUpgradeState's legacy-store business rules (a fresh backend has no
+// legacy store to upgrade from).
 async function checkReservationStoreConformance(createBackend) {
   if (typeof createBackend !== 'function') return { ok: false, errors: ['createBackend must be a factory function returning a fresh backend'] };
   const backend = createBackend();
@@ -371,6 +485,7 @@ async function checkReservationStoreConformance(createBackend) {
 
 module.exports = {
   RESERVATION_STORE_SCHEMA_VERSION,
+  RESERVATION_STORE_SCHEMA_VERSION_V1,
   RESERVATION_STATES,
   emptyState,
   validateState,

modules/jarvos-storage-janitor/test/ports.test.js

@@ -20,9 +20,12 @@ test('an external reclaim port must expose proposeDryRun() and execute()', () =>
   assert.doesNotThrow(() => assertExternalReclaimPort({ proposeDryRun: () => {}, execute: () => {} }));
 });
 
-test('a reservation port must expose reserve(), consume(), reap(), and get()', () => {
+test('a reservation port must expose reserve(), consume(), release(), reap(), and get()', () => {
   assert.throws(() => assertReservationPort({ reserve: () => {} }));
-  assert.doesNotThrow(() => assertReservationPort({
+  assert.throws(() => assertReservationPort({
     reserve: () => {}, consume: () => {}, reap: () => {}, get: () => {},
   }));
+  assert.doesNotThrow(() => assertReservationPort({
+    reserve: () => {}, consume: () => {}, release: () => {}, reap: () => {}, get: () => {},
+  }));
 });