SILVER/NODEDC_PLATFORM
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

fix(edp): resolve managed reader source scope

Browse Source
This commit is contained in:
Codex
2026-07-19 12:28:00 +03:00
parent 0611a88971
commit 95446bdc24
10 changed files with 273 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/external-data-plane/src/data-product-delivery.mjs
+22 -5
View File
@@ -187,6 +187,7 @@ export async function persistDataProductPublish(pool, batch, definition, {
}
export async function readDataProductSnapshot(pool, binding, definition, { limit = 5000 } = {}) {
const sourceConnectionId = readerSourceConnectionId(binding);
const client = await pool.connect();
try {
await client.query("begin isolation level repeatable read read only");
@@ -202,7 +203,7 @@ export async function readDataProductSnapshot(pool, binding, definition, { limit
where tenant_id = $1 and connection_id = $2 and provider_id = $3 and data_product_id = $4
order by source_id asc, semantic_type asc
limit $5`,
[binding.tenantId, binding.connectionId, binding.providerId, definition.id, limit + 1],
[binding.tenantId, sourceConnectionId, binding.providerId, definition.id, limit + 1],
);
if (rows.rowCount > limit) throw deliveryError("data_product_snapshot_limit_exceeded", 413);
await client.query("commit");
@@ -266,6 +267,7 @@ export function normalizeHistoryReadOptions(value, definition) {
}
export async function readDataProductHistory(pool, binding, definition, options) {
const sourceConnectionId = readerSourceConnectionId(binding);
const query = normalizeHistoryReadOptions(options, definition);
const cursor = query.cursor || {};
const rows = await pool.query(
@@ -302,7 +304,7 @@ export async function readDataProductHistory(pool, binding, definition, options)
order by query_bucket asc, source_id asc, semantic_type asc
limit $13`,
[
binding.tenantId, binding.connectionId, binding.providerId, definition.id,
binding.tenantId, sourceConnectionId, binding.providerId, definition.id,
query.from, query.to, query.sourceIds, query.resolutionMs,
Boolean(query.cursor), cursor.bucketStart || query.from, cursor.sourceId || "", cursor.semanticType || "",
query.limit + 1,
@@ -337,6 +339,7 @@ export async function readDataProductHistory(pool, binding, definition, options)
}
export async function readPatchEvents(pool, binding, definition, after, { limit = 100 } = {}) {
const sourceConnectionId = readerSourceConnectionId(binding);
const client = await pool.connect();
try {
await client.query("begin isolation level repeatable read read only");
@@ -351,7 +354,7 @@ export async function readPatchEvents(pool, binding, definition, after, { limit
where tenant_id = $1 and connection_id = $2 and provider_id = $3
and data_product_id = $4 and cursor > $5
order by cursor asc limit $6`,
[binding.tenantId, binding.connectionId, binding.providerId, definition.id, after.toString(), limit],
[binding.tenantId, sourceConnectionId, binding.providerId, definition.id, after.toString(), limit],
);
await client.query("commit");
return rows.rows.map((row) => ({
@@ -518,26 +521,40 @@ async function persistPatchChunks(client, batch, definition, batchId, chunks) {
async function currentCursor(db, scope, explicitProductId) {
const dataProductId = explicitProductId || scope.contract.dataProductId;
const connectionId = scope.source
? scope.source.connectionId
: readerSourceConnectionId(scope);
const result = await db.query(
`select current_cursor from external_data_plane_delivery_state
where tenant_id = $1 and connection_id = $2 and provider_id = $3 and data_product_id = $4`,
[scope.tenantId || scope.source.tenantId, scope.connectionId || scope.source.connectionId,
[scope.tenantId || scope.source.tenantId, connectionId,
scope.providerId || scope.source.providerId, dataProductId],
);
return result.rowCount ? BigInt(result.rows[0].current_cursor) : 0n;
}
async function patchFloor(db, binding, dataProductId) {
const sourceConnectionId = readerSourceConnectionId(binding);
const result = await db.query(
`select min(cursor) as floor from external_data_plane_patch_outbox
where tenant_id = $1 and connection_id = $2 and provider_id = $3 and data_product_id = $4`,
[binding.tenantId, binding.connectionId, binding.providerId, dataProductId],
[binding.tenantId, sourceConnectionId, binding.providerId, dataProductId],
);
if (result.rows[0]?.floor !== null && result.rows[0]?.floor !== undefined) return BigInt(result.rows[0].floor);
const current = await currentCursor(db, binding, dataProductId);
return current + 1n;
}
function readerSourceConnectionId(binding) {
const sourceConnectionId = String(binding?.sourceConnectionId || "").trim();
if (sourceConnectionId) return sourceConnectionId;
if (!binding?.bindingKey) {
const legacyConnectionId = String(binding?.connectionId || "").trim();
if (legacyConnectionId) return legacyConnectionId;
}
throw deliveryError("reader_source_scope_unresolved", 409);
}
function scopeValues(batch, tail) {
return [
batch.source.tenantId, batch.source.connectionId, batch.source.providerId,
services/external-data-plane/src/reader-source-scope.mjs
+87
View File
@@ -0,0 +1,87 @@
const IDENTIFIER = /^[a-z][a-z0-9._:-]{2,127}$/;
export async function resolveManagedReaderSourceConnection(db, policy) {
const tenantId = identifier(policy?.tenantId);
const consumerConnectionId = identifier(policy?.connectionId);
const providerId = identifier(policy?.providerId);
const allowedDataProductIds = uniqueIdentifiers(policy?.allowedDataProductIds);
if (!tenantId || !consumerConnectionId || !providerId || !allowedDataProductIds.length) {
throw sourceScopeError("managed_reader_source_scope_request_invalid", 400);
}
const result = await db.query(
`select distinct candidate.connection_id as "connectionId"
from external_data_plane_writer_bindings as candidate
where candidate.tenant_id = $1
and candidate.provider_id = $2
and candidate.active = true
and (candidate.expires_at is null or candidate.expires_at > now())
and not exists (
select 1
from unnest($3::text[]) as requested(data_product_id)
where not exists (
select 1
from external_data_plane_writer_bindings as coverage
where coverage.tenant_id = candidate.tenant_id
and coverage.connection_id = candidate.connection_id
and coverage.provider_id = candidate.provider_id
and coverage.active = true
and (coverage.expires_at is null or coverage.expires_at > now())
and coverage.allowed_data_product_ids ? requested.data_product_id
)
)
order by candidate.connection_id asc
limit 3`,
[tenantId, providerId, allowedDataProductIds],
);
const candidates = uniqueIdentifiers(result.rows.map((row) => row.connectionId));
if (candidates.includes(consumerConnectionId)) return consumerConnectionId;
if (candidates.length === 1) return candidates[0];
if (!candidates.length) throw sourceScopeError("managed_reader_source_scope_not_found", 409);
throw sourceScopeError("managed_reader_source_scope_ambiguous", 409);
}
export async function backfillManagedReaderSourceConnections(db) {
const unresolved = await db.query(
`select id, tenant_id as "tenantId", connection_id as "connectionId",
provider_id as "providerId", allowed_data_product_ids as "allowedDataProductIds"
from external_data_plane_reader_bindings
where binding_key is not null and source_connection_id is null
order by created_at asc, id asc`,
);
let resolved = 0;
let unresolvedCount = 0;
for (const binding of unresolved.rows) {
try {
const sourceConnectionId = await resolveManagedReaderSourceConnection(db, binding);
const result = await db.query(
`update external_data_plane_reader_bindings
set source_connection_id = $2
where id = $1 and binding_key is not null and source_connection_id is null`,
[binding.id, sourceConnectionId],
);
resolved += result.rowCount;
} catch (error) {
if (!new Set([
"managed_reader_source_scope_not_found",
"managed_reader_source_scope_ambiguous",
]).has(error?.code)) throw error;
unresolvedCount += 1;
}
}
return { resolved, unresolved: unresolvedCount };
}
function sourceScopeError(code, status) {
return Object.assign(new Error(code), { code, status });
}
function identifier(value) {
const normalized = typeof value === "string" ? value.trim() : "";
return IDENTIFIER.test(normalized) ? normalized : "";
}
function uniqueIdentifiers(value) {
if (!Array.isArray(value)) return [];
return [...new Set(value.map(identifier).filter(Boolean))];
}
services/external-data-plane/src/schema.mjs
+10
View File
@@ -1,3 +1,5 @@
import { backfillManagedReaderSourceConnections } from "./reader-source-scope.mjs";
export async function migrate(pool) {
await pool.query("create extension if not exists timescaledb");
await pool.query("create extension if not exists postgis");
@@ -160,6 +162,7 @@ export async function migrate(pool) {
await pool.query("alter table external_data_plane_reader_bindings add column if not exists binding_key text");
await pool.query("alter table external_data_plane_reader_bindings add column if not exists request_hash text");
await pool.query("alter table external_data_plane_reader_bindings add column if not exists generation integer not null default 1");
await pool.query("alter table external_data_plane_reader_bindings add column if not exists source_connection_id text");
await pool.query("alter table external_data_plane_reader_bindings alter column expires_at drop not null");
await pool.query(`
do $$
@@ -202,6 +205,13 @@ export async function migrate(pool) {
`);
await pool.query("create unique index if not exists external_data_plane_reader_bindings_managed_key_idx on external_data_plane_reader_bindings (binding_key, generation) where binding_key is not null");
await pool.query("create index if not exists external_data_plane_reader_bindings_active_idx on external_data_plane_reader_bindings (active, expires_at)");
await pool.query("create index if not exists external_data_plane_reader_bindings_source_idx on external_data_plane_reader_bindings (tenant_id, source_connection_id, provider_id, active)");
await pool.query(`
update external_data_plane_reader_bindings
set source_connection_id = connection_id
where binding_key is null and source_connection_id is null
`);
await backfillManagedReaderSourceConnections(pool);
await pool.query(`
create table if not exists external_data_plane_facts (
services/external-data-plane/src/server.mjs
+42 -10
View File
@@ -32,6 +32,7 @@ import {
readerBindingRequestHash,
safeReaderBinding,
} from "./reader-binding.mjs";
import { resolveManagedReaderSourceConnection } from "./reader-source-scope.mjs";
import { migrate } from "./schema.mjs";
import {
createWriterToken,
@@ -97,6 +98,7 @@ app.get("/healthz", asyncRoute(async (_req, res) => {
managedWriterBindings: "digest+idempotent-generation+explicit-revoke",
readerBindings: "supported",
managedReaderBindings: "digest+idempotent-generation+explicit-revoke",
readerSourceScope: "writer-resolved+fail-closed-ambiguity",
dataProductDelivery: "snapshot+history+durable-patch",
legacyIntake: config.legacyIntakeEnabled ? "migration-only" : "disabled",
legacyCapabilityProvisioning: config.provisionerApiEnabled ? "enabled" : "disabled",
@@ -388,7 +390,8 @@ app.put("/internal/data-plane/v1/reader-bindings/by-key/:bindingKey", requireMan
await client.query("select pg_advisory_xact_lock(hashtextextended($1, 0))", [bindingKey]);
const existing = await client.query(
`select id, binding_key as "bindingKey", request_hash as "requestHash", generation,
tenant_id as "tenantId", connection_id as "connectionId", provider_id as "providerId",
tenant_id as "tenantId", connection_id as "connectionId",
source_connection_id as "sourceConnectionId", provider_id as "providerId",
allowed_data_product_ids as "allowedDataProductIds", expires_at as "expiresAt",
active, created_at as "createdAt", rotated_at as "rotatedAt", revoked_at as "revokedAt"
from external_data_plane_reader_bindings
@@ -397,22 +400,40 @@ app.put("/internal/data-plane/v1/reader-bindings/by-key/:bindingKey", requireMan
[bindingKey, policy.generation],
);
if (existing.rowCount) {
const binding = existing.rows[0];
let binding = existing.rows[0];
if (binding.requestHash !== requestHash) {
throw httpError(409, "managed_reader_binding_request_conflict");
}
if (binding.active !== true || binding.expiresAt !== null) {
throw httpError(409, "managed_reader_binding_generation_inactive");
}
if (!binding.sourceConnectionId) {
const sourceConnectionId = await resolveManagedReaderSourceConnection(client, policy);
const resolved = await client.query(
`update external_data_plane_reader_bindings
set source_connection_id = $3
where binding_key = $1 and generation = $2 and source_connection_id is null
returning id, binding_key as "bindingKey", request_hash as "requestHash", generation,
tenant_id as "tenantId", connection_id as "connectionId",
source_connection_id as "sourceConnectionId", provider_id as "providerId",
allowed_data_product_ids as "allowedDataProductIds", expires_at as "expiresAt",
active, created_at as "createdAt", rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
[bindingKey, policy.generation, sourceConnectionId],
);
binding = resolved.rows[0] || binding;
}
response = { status: 200, idempotent: true, binding };
} else {
const sourceConnectionId = await resolveManagedReaderSourceConnection(client, policy);
const inserted = await client.query(
`insert into external_data_plane_reader_bindings (
id, token_hash, binding_key, request_hash, generation,
tenant_id, connection_id, provider_id, allowed_data_product_ids, expires_at
) values ($1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb, $10)
tenant_id, connection_id, source_connection_id, provider_id,
allowed_data_product_ids, expires_at
) values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10::jsonb, $11)
returning id, binding_key as "bindingKey", generation,
tenant_id as "tenantId", connection_id as "connectionId", provider_id as "providerId",
tenant_id as "tenantId", connection_id as "connectionId",
source_connection_id as "sourceConnectionId", provider_id as "providerId",
allowed_data_product_ids as "allowedDataProductIds", expires_at as "expiresAt",
active, created_at as "createdAt", rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
[
@@ -423,6 +444,7 @@ app.put("/internal/data-plane/v1/reader-bindings/by-key/:bindingKey", requireMan
policy.generation,
policy.tenantId,
policy.connectionId,
sourceConnectionId,
policy.providerId,
JSON.stringify(policy.allowedDataProductIds),
policy.expiresAt,
@@ -459,7 +481,8 @@ app.post("/internal/data-plane/v1/reader-bindings/by-key/:bindingKey/generations
await client.query("select pg_advisory_xact_lock(hashtextextended($1, 0))", [bindingKey]);
const existing = await client.query(
`select id, binding_key as "bindingKey", generation,
tenant_id as "tenantId", connection_id as "connectionId", provider_id as "providerId",
tenant_id as "tenantId", connection_id as "connectionId",
source_connection_id as "sourceConnectionId", provider_id as "providerId",
allowed_data_product_ids as "allowedDataProductIds", expires_at as "expiresAt",
active, created_at as "createdAt", rotated_at as "rotatedAt", revoked_at as "revokedAt"
from external_data_plane_reader_bindings
@@ -474,7 +497,8 @@ app.post("/internal/data-plane/v1/reader-bindings/by-key/:bindingKey/generations
set active = false, revoked_at = now()
where binding_key = $1 and generation = $2 and active = true
returning id, binding_key as "bindingKey", generation,
tenant_id as "tenantId", connection_id as "connectionId", provider_id as "providerId",
tenant_id as "tenantId", connection_id as "connectionId",
source_connection_id as "sourceConnectionId", provider_id as "providerId",
allowed_data_product_ids as "allowedDataProductIds", expires_at as "expiresAt",
active, created_at as "createdAt", rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
[bindingKey, generation],
@@ -505,10 +529,11 @@ app.post("/internal/data-plane/v1/reader-bindings", requireProvisionerApi, async
const bindingId = randomUUID();
const result = await pool.query(
`insert into external_data_plane_reader_bindings (
id, token_hash, tenant_id, connection_id, provider_id,
id, token_hash, tenant_id, connection_id, source_connection_id, provider_id,
allowed_data_product_ids, expires_at
) values ($1, $2, $3, $4, $5, $6::jsonb, $7)
) values ($1, $2, $3, $4, $4, $5, $6::jsonb, $7)
returning id, tenant_id as "tenantId", connection_id as "connectionId",
source_connection_id as "sourceConnectionId",
provider_id as "providerId", allowed_data_product_ids as "allowedDataProductIds",
expires_at as "expiresAt", active, created_at as "createdAt",
rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
@@ -533,6 +558,7 @@ app.post("/internal/data-plane/v1/reader-bindings/:bindingId/rotate", requirePro
set token_hash = $2, rotated_at = now()
where id = $1 and binding_key is null and active = true and expires_at > now()
returning id, tenant_id as "tenantId", connection_id as "connectionId",
source_connection_id as "sourceConnectionId",
provider_id as "providerId", allowed_data_product_ids as "allowedDataProductIds",
expires_at as "expiresAt", active, created_at as "createdAt",
rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
@@ -549,6 +575,7 @@ app.post("/internal/data-plane/v1/reader-bindings/:bindingId/revoke", requirePro
set active = false, revoked_at = now()
where id = $1 and binding_key is null and active = true
returning id, tenant_id as "tenantId", connection_id as "connectionId",
source_connection_id as "sourceConnectionId",
provider_id as "providerId", allowed_data_product_ids as "allowedDataProductIds",
expires_at as "expiresAt", active, created_at as "createdAt",
rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
@@ -937,7 +964,9 @@ async function resolveReaderBinding(req) {
const token = bearerToken(req);
if (!token) throw httpError(401, "reader_binding_unauthorized");
const result = await pool.query(
`select id, token_hash as "tokenHash", tenant_id as "tenantId", connection_id as "connectionId",
`select id, binding_key as "bindingKey", token_hash as "tokenHash",
tenant_id as "tenantId", connection_id as "connectionId",
source_connection_id as "sourceConnectionId",
provider_id as "providerId", allowed_data_product_ids as "allowedDataProductIds",
expires_at as "expiresAt", active, created_at as "createdAt",
rotated_at as "rotatedAt", revoked_at as "revokedAt"
@@ -947,6 +976,9 @@ async function resolveReaderBinding(req) {
[hashReaderToken(token)],
);
if (!result.rowCount) throw httpError(401, "reader_binding_unauthorized");
if (result.rows[0].bindingKey && !result.rows[0].sourceConnectionId) {
throw httpError(409, "reader_source_scope_unresolved");
}
return result.rows[0];
}