Skip to content

Replay retained logs on remote enrollment #1034

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
platypii wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Replay retained logs on remote enrollment #1034

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
58 changes: 42 additions & 16 deletions hypaware-core/plugins-workspace/central/index.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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'

/**
Expand Down Expand Up @@ -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,
Expand Down
12 changes: 9 additions & 3 deletions hypaware-core/plugins-workspace/central/proto.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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`
Expand All @@ -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.
Expand Down
56 changes: 48 additions & 8 deletions hypaware-core/plugins-workspace/central/src/identity_client.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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'
*/

/**
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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)
}
Expand All @@ -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 }
}
}

/**
Expand All @@ -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`)
Expand All @@ -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
Expand Down Expand Up @@ -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 }
}

/**
Expand Down
Loading
Loading