Replay retained logs on remote enrollment

Author: platypii

hypaware-core/plugins-workspace/central/index.js

@@ -2,12 +2,12 @@
 
 import path from 'node:path'
 
-import { createInstanceWatermarkStore } from '../../../src/core/sinks/incremental.js'
+import { createSinkWatermarkStore } from '../../../src/core/sinks/watermarks.js'
 
 import { validateCentralConfig } from './src/config.js'
 import { createConfigPullLoop } from './src/config_client.js'
 import { IdentityClient } from './src/identity_client.js'
-import { createDatasetRolloutStore } from './src/rollout.js'
+import { bindDestinationState, createDatasetRolloutStore, markDestinationStateReady } from './src/rollout.js'
 import { createForwardSink, initializeOpenDatasetRollouts } from './src/sink.js'
 
 /**
@@ -63,22 +63,48 @@ export async function activate(ctx) {
         hyp_identity_source: source,
       })
 
-      // Per-(sink instance, partition) incremental-read watermarks. The plugin
-      // `stateDir` is per-PLUGIN, so two `@hypaware/central` instances would
-      // share, and clobber, one watermark file and skip each other's rows;
-      // `createInstanceWatermarkStore` namespaces by the instance name, matching
-      // local-fs/s3. Each forward instance then reads only rows added since its
-      // own last successful export.
-      // @ref LLP 0040#watermark-contract [implements]: one watermark per (sink instance, partition), scoped by instance name
-      const watermarks = createInstanceWatermarkStore({ paths: sinkCtx.paths, instanceName: sinkCtx.name })
-      const rollouts = createDatasetRolloutStore({ paths: sinkCtx.paths, instanceName: sinkCtx.name })
+      // Bind progress before creating either state store. Existing unscoped
+      // progress is adopted once for the current destination; a new origin/org
+      // gets an isolated scope durably marked for retained-history replay.
+      // @ref LLP 0315#destination-identity [implements]: watermarks and rollout manifests share one destination-scoped state root
+      let destinationState = await bindDestinationState({
+        paths: sinkCtx.paths,
+        instanceName: sinkCtx.name,
+        destination: identityClient.getDestination(),
+      })
+      sinkCtx.log.info('central.destination.bound', {
+        hyp_sink_instance: sinkCtx.name,
+        destination_origin: destinationState.destination.origin,
+        destination_org: destinationState.destination.org,
+        destination_phase: destinationState.phase,
+        adopted_legacy_progress: destinationState.adoptedLegacy,
+      })
 
-      // Establish open-dataset rollout state during sink creation. On an
-      // upgraded machine this baselines partitions already on disk; on a cold
-      // machine it durably records an empty dataset before the first captured
-      // row can be mistaken for rollout history.
+      const watermarks = createSinkWatermarkStore({ stateDir: destinationState.stateDir })
+      const rollouts = createDatasetRolloutStore({ stateDir: destinationState.stateDir })
+
+      // Establish open-dataset rollout state during sink creation. An existing
+      // destination's software rollout baselines current partitions; a new
+      // destination starts them at zero so retained eligible history forwards.
+      // An empty dataset still gets a durable manifest before its first row.
       // @ref LLP 0307#rollout-instant [implements]: initialize dataset rollout state before scheduled exports can observe a first partition
-      await initializeOpenDatasetRollouts({ query, storage, watermarks, rollouts, log: sinkCtx.log })
+      // @ref LLP 0315#new-destination-replay [implements]: a newly bound destination initializes eligible open datasets for retained-history replay
+      await initializeOpenDatasetRollouts({
+        query,
+        storage,
+        watermarks,
+        rollouts,
+        log: sinkCtx.log,
+        replayRetainedHistory: destinationState.phase === 'initializing-history',
+      })
+      if (destinationState.phase === 'initializing-history') {
+        destinationState = await markDestinationStateReady(destinationState)
+        sinkCtx.log.info('central.destination.ready', {
+          hyp_sink_instance: sinkCtx.name,
+          destination_origin: destinationState.destination.origin,
+          destination_org: destinationState.destination.org,
+        })
+      }
 
       const sink = createForwardSink({
         config,

hypaware-core/plugins-workspace/central/proto.md

@@ -24,7 +24,7 @@ release that drops `/v1/` ships. Clients send no version header.
 
 The gateway holds one long-lived JWT issued by the central server. The
 JWT's `sub` is the gateway id; the kernel persists `{ jwt, expires_at,
-gateway_id }` to `<plugin.stateDir>/identity.json` (mode 0600,
+gateway_id, org }` to `<plugin.stateDir>/identity.json` (mode 0600,
 atomic tmp+rename).
 
 ### POST `/v1/identity/bootstrap`
@@ -43,10 +43,16 @@ Request:
 Response 200:
 
 ```json
-{ "jwt": "<base64url.signed.jwt>", "expires_at": 1814400000 }
+{
+  "jwt": "<base64url.signed.jwt>",
+  "expires_at": 1814400000,
+  "org": "acme.example"
+}
 ```
 
-`expires_at` is a Unix epoch second.
+`expires_at` is a Unix epoch second. `org` is the stable server-assigned
+organization identifier used with the server origin to scope export progress;
+it is an empty string for a single-org server without organization enforcement.
 
 Response 401 / 4xx: `{ "error": "<kind>" }`. Gateway aborts; operator
 must issue a new bootstrap token.

hypaware-core/plugins-workspace/central/src/identity_client.js

@@ -5,7 +5,7 @@ import fs from 'node:fs'
 import { atomicWriteJsonSync, isPlainObject, sha256Hex } from 'hypaware/core/util'
 
 /**
- * @import { AcquireSource, PersistedIdentity } from './types.js'
+ * @import { AcquireSource, PersistedIdentity, RemoteDestination } from './types.js'
  */
 
 /**
@@ -110,6 +110,14 @@ export class IdentityClient {
         )
       }
       this.identity = persisted
+      // An identity written before destination-scoped progress has no stable
+      // organization beside it. Refresh once against the upgraded server
+      // before any sink state is selected, then persist the authoritative org.
+      // @ref LLP 0315#destination-identity [implements]: legacy identities acquire the server-assigned org before export progress is bound to a destination
+      if (persisted.org === undefined) {
+        await this.refresh()
+        return 'refreshed'
+      }
       const remainingSec = persisted.expires_at - Math.floor(this.now() / 1000)
       if (remainingSec <= REFRESH_WINDOW_SECONDS) {
         await this.refresh()
@@ -200,13 +208,23 @@ export class IdentityClient {
       throw new Error(`identity refresh failed: ${await readErrorDetail(response)}`)
     }
     const parsed = await readJsonResponse(response, 'refresh')
-    const identity = identityFromPayload(parsed, this.identity.gateway_id)
+    const previous = this.identity
+    const identity = identityFromPayload(parsed, previous.gateway_id)
+    // A credential rotation must not silently move a running sink to another
+    // destination. A legacy identity has no prior org and adopts the response;
+    // every subsequent refresh must preserve it exactly.
+    // @ref LLP 0315#destination-identity [constrained-by]: credential refresh preserves destination identity; an org change requires a new enrollment and state scope
+    if (previous.org !== undefined && identity.org !== previous.org) {
+      throw new Error(
+        `identity refresh failed: central server changed organization from '${previous.org}' to '${identity.org}'`
+      )
+    }
     // Preserve the mint provenance across refresh; the bootstrap token is
     // typically absent in steady state, so re-derive it from the prior
     // persisted identity rather than recomputing.
-    identity.central_url = this.identity.central_url
-    identity.bootstrap_token_fp = this.identity.bootstrap_token_fp
-    if (this.identity.origin !== undefined) identity.origin = this.identity.origin
+    identity.central_url = previous.central_url
+    identity.bootstrap_token_fp = previous.bootstrap_token_fp
+    if (previous.origin !== undefined) identity.origin = previous.origin
     this.identity = identity
     writePersistedFile(this.persistedPath, identity)
   }
@@ -230,6 +248,21 @@ export class IdentityClient {
     }
     return this.identity.jwt
   }
+
+  /**
+   * Return the authenticated destination identity used to scope export state.
+   * `acquire()` must run first so a legacy identity has already refreshed its
+   * missing organization.
+   *
+   * @returns {RemoteDestination}
+   */
+  getDestination() {
+    if (!this.identity || this.identity.org === undefined) {
+      throw new Error('identity destination not acquired - call acquire() first')
+    }
+    // @ref LLP 0315#destination-identity [implements]: progress keys on canonical server origin plus stable organization, never gateway credentials
+    return { origin: new URL(this.centralUrl).origin, org: this.identity.org }
+  }
 }
 
 /**
@@ -255,7 +288,7 @@ function readPersistedFile(filePath) {
   if (!isPlainObject(parsed)) {
     throw new Error(`persisted identity ${filePath} must be an object`)
   }
-  const { jwt, expires_at, gateway_id, central_url, bootstrap_token_fp, origin } =
+  const { jwt, expires_at, gateway_id, org, central_url, bootstrap_token_fp, origin } =
     /** @type {Record<string, unknown>} */ (parsed)
   if (typeof jwt !== 'string' || jwt.length === 0) {
     throw new Error(`persisted identity ${filePath}: missing or invalid jwt`)
@@ -268,6 +301,10 @@ function readPersistedFile(filePath) {
   }
   /** @type {PersistedIdentity} */
   const identity = { jwt, expires_at, gateway_id }
+  if (org !== undefined && typeof org !== 'string') {
+    throw new Error(`persisted identity ${filePath}: invalid org`)
+  }
+  if (typeof org === 'string') identity.org = org
   if (typeof central_url === 'string') identity.central_url = central_url
   if (typeof bootstrap_token_fp === 'string') identity.bootstrap_token_fp = bootstrap_token_fp
   if (origin === 'login') identity.origin = origin
@@ -328,18 +365,21 @@ function identityFromPayload(parsed, fallbackGatewayId) {
   if (!isPlainObject(parsed)) {
     throw new Error('central server response is not an object')
   }
-  const { jwt, expires_at } = /** @type {Record<string, unknown>} */ (parsed)
+  const { jwt, expires_at, org } = /** @type {Record<string, unknown>} */ (parsed)
   if (typeof jwt !== 'string' || jwt.length === 0) {
     throw new Error('central server response missing jwt')
   }
   if (typeof expires_at !== 'number' || !Number.isInteger(expires_at)) {
     throw new Error('central server response missing expires_at')
   }
+  if (typeof org !== 'string') {
+    throw new Error('central server response missing org')
+  }
   const gateway_id = decodeJwtSub(jwt) ?? fallbackGatewayId
   if (typeof gateway_id !== 'string' || gateway_id.length === 0) {
     throw new Error('central server response missing gateway identity (sub claim)')
   }
-  return { jwt, expires_at, gateway_id }
+  return { jwt, expires_at, gateway_id, org }
 }
 
 /**