mirror of
https://github.com/langgenius/dify.git
synced 2026-09-08 02:43:49 +08:00
feat(knowledge-fs): support source config updates and deferred sync
- allow updating mutable source configuration fields - enqueue sync after active business configuration changes - pause scheduled sync while a source is disabled - preserve and rebind sync policies across source version updates - support website, online document, and online drive sources
This commit is contained in:
FFXN
parent
84c53bdfde
commit
52df787444
@ -2430,12 +2430,14 @@ class KnowledgeFSSpaceSourceApi(Resource):
|
||||
@_knowledge_fs_errors
|
||||
def patch(self, control_space_id: str, source_id: str):
|
||||
actor_id, tenant_id = _actor()
|
||||
payload = _payload(KnowledgeFSSourceUpdatePayload)
|
||||
payload.sync_after_update = True
|
||||
result = _console_services().facade.update_source(
|
||||
tenant_id=tenant_id,
|
||||
account_id=actor_id,
|
||||
control_space_id=control_space_id,
|
||||
source_id=source_id,
|
||||
payload=_payload(KnowledgeFSSourceUpdatePayload),
|
||||
payload=payload,
|
||||
)
|
||||
return dump_response(KnowledgeFSSourceResponse, result)
|
||||
|
||||
|
||||
@ -1,7 +1,7 @@
|
||||
{
|
||||
"schemaVersion": 5,
|
||||
"subtreeTree": "602f9fdb2f415093507850074b71e22571b1a4cd",
|
||||
"openapiSha256": "b199b523299eff504a174bf92f8d41ab47622bef1ff30b285d49f10c0119058a",
|
||||
"subtreeTree": "151d5e5d9883f06e49be6f562b32b192b2619edc",
|
||||
"openapiSha256": "94b9676d6a2b27b79f7b2c42ca91902d3777a99b1e7efc5747f4a0fce6df8f91",
|
||||
"capabilityV2AuthManifestSha256": "e322a2fa779d1f40b95c54c1021cffecaec77abbd7b34899573dcdf4ff353109",
|
||||
"capabilityV2AuthTestVectorSha256": "ae0de37b1ff05c40f905cf17a7b410d8971acacf64db07d5ee3d6fecfa559ce3",
|
||||
"productOperationManifestSha256": "32f305d86525f31eb07a6aef1f72d286d53991f1f8ed1f9d6e52d37cf4561116",
|
||||
|
||||
@ -2159,6 +2159,7 @@ class KnowledgeFSSourceUpdatePayload(BaseModel):
|
||||
provider_parameters: dict[str, bool | FiniteFloat | str] | None = Field(
|
||||
default=None, max_length=50, alias="providerParameters"
|
||||
)
|
||||
sync_after_update: bool | None = Field(default=None, alias="syncAfterUpdate")
|
||||
status: Literal["active", "disabled", "error", "syncing"] | None = None
|
||||
uri: str | None = Field(default=None, min_length=1, max_length=4_096)
|
||||
|
||||
|
||||
@ -920,6 +920,17 @@ def test_source_update_payload_accepts_mutable_source_configuration() -> None:
|
||||
}
|
||||
|
||||
|
||||
def test_source_update_payload_serializes_backend_sync_admission_flag() -> None:
|
||||
payload = KnowledgeFSSourceUpdatePayload.model_validate(
|
||||
{"name": "Renamed", "syncAfterUpdate": True}
|
||||
)
|
||||
|
||||
assert payload.model_dump(mode="json", by_alias=True, exclude_none=True) == {
|
||||
"name": "Renamed",
|
||||
"syncAfterUpdate": True,
|
||||
}
|
||||
|
||||
|
||||
def test_source_update_payload_requires_provider_parameters_for_parameter_updates() -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
KnowledgeFSSourceUpdatePayload.model_validate({"metadata": {"parameters": {"limit": 50}}})
|
||||
|
||||
@ -743,7 +743,7 @@ currently visible.
|
||||
> **Source object & conventions**
|
||||
> - `Source` = `{ id, knowledgeSpaceId, name, type: enum(upload|object-storage|connector|web), uri, status: enum(active|syncing|error|disabled), permissionScope: string[], metadata, version: int≥1, createdAt, updatedAt }`.
|
||||
> - **Optimistic locking**: `version` is bumped on every write. Pass it back as `expectedVersion` on PATCH to fail with `409` instead of overwriting a concurrent modification.
|
||||
> - **Scheduled sync**: set `metadata.syncPolicy` to `{"everyHours": N}` (1–720) or `{"dailyAt": ["HH:MM", …], "utcOffset": "±HH:MM"?}`. Invalid policies are rejected with `400` at create/update. A background scheduler (multi-replica safe; `KNOWLEDGE_SOURCE_SYNC*` env) then re-syncs due sources: web → re-crawl with content-hash dedup, connector pages → re-import previously imported pages whose `lastEditedTime` changed, drive → re-download previously imported files. The scheduler records progress under `metadata.syncState` `{ lastSyncAt, lastSyncStatus: ok|error, lastSyncError?, nextSyncAt, syncStartedAt? }`.
|
||||
> - **Scheduled sync**: set `metadata.syncPolicy` to `{"everyHours": N}` (1–720) or `{"dailyAt": ["HH:MM", …], "utcOffset": "±HH:MM"?}`. Invalid policies are rejected with `400` at create/update. A background scheduler (multi-replica safe; `KNOWLEDGE_SOURCE_SYNC*` env) then re-syncs due sources: web → re-crawl with content-hash dedup, connector pages → re-import previously imported pages whose `lastEditedTime` changed, drive → re-download previously imported files. A disabled Source pauses due scheduling without disabling or advancing its policy; after re-enabling, an overdue policy is eligible on the next scheduler pass. Source updates rebind the unchanged policy to the new Source version. The scheduler records progress under `metadata.syncState` `{ lastSyncAt, lastSyncStatus: ok|error, lastSyncError?, nextSyncAt, syncStartedAt? }`.
|
||||
> - **Reserved metadata keys** (managed by the platform; user PATCHes should carry them through): `tenantId` (auto-stamped on create/update), `sync` (last sync summary), `crawled`, `imported`, `importedFiles` (per-source sync state), `syncState`.
|
||||
|
||||
### `POST /knowledge-spaces/{id}/sources`
|
||||
@ -769,9 +769,9 @@ currently visible.
|
||||
### `PATCH /knowledge-spaces/{id}/sources/{sourceId}`
|
||||
**Description**: Update a source's mutable fields, optionally guarded by optimistic locking.
|
||||
**Auth**: Bearer; scope `knowledge-spaces:write`.
|
||||
**Path params**: `id` (uuid); `sourceId` (uuid). **Body** (`application/json`, strict; at least one mutable field is required): `name` (1–200); `status` (enum); `metadata` (object, recursively patches ordinary metadata and cannot contain `parameters`; `tenantId` is re-stamped automatically); `providerParameters` (object containing at most 50 boolean/finite-number/string values, fully replaces `metadata.parameters`, so omitted keys are removed); `uri` (1–4096); `expectedVersion` (int ≥1) — the `version` from your last read; when provided, a concurrent modification makes the update fail with `409` instead of overwriting it. For web sources, changing `uri` also changes the provider parameter `url`, while a replacement `providerParameters.url` changes `uri`; if both are supplied, `uri` is authoritative.
|
||||
**Path params**: `id` (uuid); `sourceId` (uuid). **Body** (`application/json`, strict; at least one mutable field is required): `name` (1–200); `status` (enum); `metadata` (object, recursively patches ordinary metadata and cannot contain `parameters`; `tenantId` is re-stamped automatically); `providerParameters` (object containing at most 50 boolean/finite-number/string values, fully replaces `metadata.parameters`, so omitted keys are removed); `uri` (1–4096); `expectedVersion` (int ≥1) — the `version` from your last read; when provided, a concurrent modification makes the update fail with `409` instead of overwriting it; `syncAfterUpdate` (boolean, optional) asks the product gateway to enqueue a durable sync only when URI, provider parameters, or other business metadata actually changed. Name-only and `metadata.syncPolicy`-only updates do not enqueue a sync. For web sources, changing `uri` also changes the provider parameter `url`, while a replacement `providerParameters.url` changes `uri`; if both are supplied, `uri` is authoritative. Source identity metadata (`datasource`, `pluginId`, `provider`, `providerId`, and `providerKind`) is immutable.
|
||||
**Responses**:
|
||||
- `200`: updated `Source` (`version` bumped).
|
||||
- `200`: updated `Source` (`version` bumped), optionally enriched with `syncWorkflow` when a configuration sync was accepted.
|
||||
- `400` invalid `metadata.syncPolicy`, provider parameters under `metadata`, credential-shaped provider parameters, or an invalid web Source URL; `404`; `409` concurrent modification (`expectedVersion` mismatch) or protected legacy credential-bearing parameters; `401`/`403`.
|
||||
|
||||
### `DELETE /knowledge-spaces/{id}/sources/{sourceId}`
|
||||
|
||||
@ -393,6 +393,29 @@ describe("knowledge space source CRUD", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it("rejects source identity changes through metadata patches", async () => {
|
||||
const app = createApp();
|
||||
const spaceId = await createSpace(app);
|
||||
const sourceId = await createWebSource(app, spaceId);
|
||||
|
||||
for (const [key, value] of [
|
||||
["datasource", "other-datasource"],
|
||||
["pluginId", "other/plugin"],
|
||||
["provider", "other-provider"],
|
||||
["providerId", "other-provider-id"],
|
||||
["providerKind", "online-drive"],
|
||||
] as const) {
|
||||
const response = await app.request(`/knowledge-spaces/${spaceId}/sources/${sourceId}`, {
|
||||
body: JSON.stringify({ metadata: { [key]: value } }),
|
||||
headers: json(writeToken),
|
||||
method: "PATCH",
|
||||
});
|
||||
|
||||
expect(response.status).toBe(400);
|
||||
await expect(response.json()).resolves.toEqual({ error: `Source ${key} cannot be changed` });
|
||||
}
|
||||
});
|
||||
|
||||
it("rejects credentials disguised as provider parameters", async () => {
|
||||
const app = createApp();
|
||||
const spaceId = await createSpace(app);
|
||||
|
||||
@ -2174,6 +2174,8 @@ export function createKnowledgeGateway({
|
||||
...(sourceCredentials ? { sourceCredentials } : {}),
|
||||
...(sourceProduct ? { sourceConnections: sourceProduct.connections } : {}),
|
||||
...(sourceProduct ? { sourceProductWorkflows: sourceProduct.repository } : {}),
|
||||
...(sourceProductWorkflows ? { sourceProductWorkflowService: sourceProductWorkflows } : {}),
|
||||
...(sourceProduct ? { sourceSyncPolicyVersionBinder: sourceProduct.repository } : {}),
|
||||
legacyMutationEndpointsEnabled: sourceProduct === undefined,
|
||||
sourceDocumentMaterializer,
|
||||
sources: sourceRepository,
|
||||
|
||||
@ -16,6 +16,7 @@ import {
|
||||
type SourceCredentialTester,
|
||||
type SourceDocumentMaterializer,
|
||||
type SourceProductWorkflowRepository,
|
||||
type SourceProductWorkflowService,
|
||||
type SourceRepository,
|
||||
type SourceSecretStore,
|
||||
type WebsiteCrawlConnector,
|
||||
@ -1420,6 +1421,11 @@ describe("source handlers without optional collaborators", () => {
|
||||
SourceProductWorkflowRepository,
|
||||
"listLatestSyncCompletions" | "listLatestSyncRuns" | "listSyncPolicies"
|
||||
>;
|
||||
sourceProductWorkflowService?: Pick<SourceProductWorkflowService, "createSync">;
|
||||
sourceSyncPolicyVersionBinder?: Pick<
|
||||
SourceProductWorkflowRepository,
|
||||
"rebindSyncPolicySourceVersion"
|
||||
>;
|
||||
sources?: SourceRepository;
|
||||
websiteCrawlConnector?: WebsiteCrawlConnector;
|
||||
}
|
||||
@ -1472,6 +1478,12 @@ describe("source handlers without optional collaborators", () => {
|
||||
...(options.sourceProductWorkflows
|
||||
? { sourceProductWorkflows: options.sourceProductWorkflows }
|
||||
: {}),
|
||||
...(options.sourceProductWorkflowService
|
||||
? { sourceProductWorkflowService: options.sourceProductWorkflowService }
|
||||
: {}),
|
||||
...(options.sourceSyncPolicyVersionBinder
|
||||
? { sourceSyncPolicyVersionBinder: options.sourceSyncPolicyVersionBinder }
|
||||
: {}),
|
||||
sources,
|
||||
spaces,
|
||||
...(options.websiteCrawlConnector
|
||||
@ -1501,6 +1513,139 @@ describe("source handlers without optional collaborators", () => {
|
||||
return { sourceId: source.id, spaceId: space.id };
|
||||
}
|
||||
|
||||
it.each([
|
||||
["website", "web", { uri: "https://docs.example.com" }],
|
||||
["online document", "connector", { providerParameters: { database: "docs" } }],
|
||||
["online drive", "connector", { providerParameters: { folder: "reports" } }],
|
||||
] as const)(
|
||||
"enqueues a durable sync after a %s configuration update",
|
||||
async (_kind, sourceType, update) => {
|
||||
const createSync = vi.fn(async ({ knowledgeSpaceId, sourceId, idempotencyKey }) => ({
|
||||
checkpoint: "queued" as const,
|
||||
createdAt: "2026-08-30T00:00:00.000Z",
|
||||
executionAttempts: 0,
|
||||
id: "00000000-0000-4000-8000-000000000501",
|
||||
idempotencyKey,
|
||||
knowledgeSpaceId,
|
||||
kind: "sync" as const,
|
||||
maxExecutionAttempts: 5,
|
||||
payload: {},
|
||||
progressCompleted: 0,
|
||||
progressFailed: 0,
|
||||
progressSkipped: 0,
|
||||
rowVersion: 1,
|
||||
sourceId,
|
||||
state: "queued" as const,
|
||||
tenantId: "tenant-1",
|
||||
updatedAt: "2026-08-30T00:00:00.000Z",
|
||||
}));
|
||||
const bare = createBareApp({ sourceProductWorkflowService: { createSync } });
|
||||
const { sourceId, spaceId } = await seedSource(bare, sourceType);
|
||||
|
||||
const response = await bare.app.request(`/knowledge-spaces/${spaceId}/sources/${sourceId}`, {
|
||||
body: JSON.stringify({ ...update, syncAfterUpdate: true }),
|
||||
headers: { "content-type": "application/json" },
|
||||
method: "PATCH",
|
||||
});
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
await expect(response.json()).resolves.toMatchObject({
|
||||
id: sourceId,
|
||||
syncWorkflow: { kind: "sync", sourceId, state: "queued" },
|
||||
});
|
||||
expect(createSync).toHaveBeenCalledOnce();
|
||||
expect(createSync).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
idempotencyKey: `source-config-update:${sourceId}:2`,
|
||||
knowledgeSpaceId: spaceId,
|
||||
sourceId,
|
||||
}),
|
||||
);
|
||||
},
|
||||
);
|
||||
|
||||
it.each([
|
||||
["website", "web", { uri: "https://docs.example.com" }],
|
||||
["online document", "connector", { providerParameters: { database: "docs" } }],
|
||||
["online drive", "connector", { providerParameters: { folder: "reports" } }],
|
||||
] as const)(
|
||||
"rebinds an existing sync policy after a disabled %s configuration update without syncing",
|
||||
async (_kind, sourceType, update) => {
|
||||
const createSync = vi.fn();
|
||||
const rebindSyncPolicySourceVersion = vi.fn(async () => null);
|
||||
const bare = createBareApp({
|
||||
sourceProductWorkflowService: { createSync },
|
||||
sourceSyncPolicyVersionBinder: { rebindSyncPolicySourceVersion },
|
||||
});
|
||||
const { sourceId, spaceId } = await seedSource(bare, sourceType);
|
||||
|
||||
const disable = await bare.app.request(`/knowledge-spaces/${spaceId}/sources/${sourceId}`, {
|
||||
body: JSON.stringify({ status: "disabled", syncAfterUpdate: true }),
|
||||
headers: { "content-type": "application/json" },
|
||||
method: "PATCH",
|
||||
});
|
||||
expect(disable.status).toBe(200);
|
||||
|
||||
const updateWhileDisabled = await bare.app.request(
|
||||
`/knowledge-spaces/${spaceId}/sources/${sourceId}`,
|
||||
{
|
||||
body: JSON.stringify({ ...update, syncAfterUpdate: true }),
|
||||
headers: { "content-type": "application/json" },
|
||||
method: "PATCH",
|
||||
},
|
||||
);
|
||||
expect(updateWhileDisabled.status).toBe(200);
|
||||
expect(createSync).not.toHaveBeenCalled();
|
||||
expect(rebindSyncPolicySourceVersion).toHaveBeenNthCalledWith(1, {
|
||||
expectedSourceVersion: 1,
|
||||
knowledgeSpaceId: spaceId,
|
||||
sourceId,
|
||||
sourceVersion: 2,
|
||||
tenantId: "tenant-1",
|
||||
});
|
||||
expect(rebindSyncPolicySourceVersion).toHaveBeenNthCalledWith(2, {
|
||||
expectedSourceVersion: 2,
|
||||
knowledgeSpaceId: spaceId,
|
||||
sourceId,
|
||||
sourceVersion: 3,
|
||||
tenantId: "tenant-1",
|
||||
});
|
||||
},
|
||||
);
|
||||
|
||||
it.each([
|
||||
["name", { name: "Renamed" }],
|
||||
["sync policy", { metadata: { syncPolicy: { everyHours: 24 } } }],
|
||||
] as const)("does not enqueue a sync for a %s-only update", async (_kind, update) => {
|
||||
const createSync = vi.fn();
|
||||
const bare = createBareApp({ sourceProductWorkflowService: { createSync } });
|
||||
const { sourceId, spaceId } = await seedSource(bare, "web");
|
||||
|
||||
const response = await bare.app.request(`/knowledge-spaces/${spaceId}/sources/${sourceId}`, {
|
||||
body: JSON.stringify({ ...update, syncAfterUpdate: true }),
|
||||
headers: { "content-type": "application/json" },
|
||||
method: "PATCH",
|
||||
});
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
expect(createSync).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("does not enqueue a sync when submitted configuration is unchanged", async () => {
|
||||
const createSync = vi.fn();
|
||||
const bare = createBareApp({ sourceProductWorkflowService: { createSync } });
|
||||
const { sourceId, spaceId } = await seedSource(bare, "web");
|
||||
|
||||
const response = await bare.app.request(`/knowledge-spaces/${spaceId}/sources/${sourceId}`, {
|
||||
body: JSON.stringify({ syncAfterUpdate: true, uri: "https://example.com" }),
|
||||
headers: { "content-type": "application/json" },
|
||||
method: "PATCH",
|
||||
});
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
expect(createSync).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("enriches a source list with product sync details in bulk lookups", async () => {
|
||||
const listSyncPolicies = vi.fn();
|
||||
const listLatestSyncCompletions = vi.fn();
|
||||
|
||||
@ -1,3 +1,4 @@
|
||||
import { isDeepStrictEqual } from "node:util";
|
||||
import type { OpenAPIHono } from "@hono/zod-openapi";
|
||||
|
||||
import {
|
||||
@ -20,6 +21,7 @@ import { type KnowledgeFsPublicFailure, knowledgeFsFailureForCode } from "./know
|
||||
import type { KnowledgeSpaceRepository } from "./knowledge-space-repository";
|
||||
import type { OnlineDocumentConnector } from "./online-document-connector";
|
||||
import type { OnlineDriveConnector } from "./online-drive-connector";
|
||||
import type { LooseOpenApiContext } from "./openapi-handler-utils";
|
||||
import type { SourceConnectionService } from "./source-connection";
|
||||
import {
|
||||
SOURCE_DOCUMENT_REPLACEMENT_SAGA_REQUIRED,
|
||||
@ -41,6 +43,7 @@ import type {
|
||||
import { safeSourceOperationError, sourceOperationFailureMetadata } from "./source-operation-error";
|
||||
import {
|
||||
type SourceProductWorkflowRepository,
|
||||
type SourceProductWorkflowService,
|
||||
type SourceWorkflowRun,
|
||||
toPublicSourceWorkflowRun,
|
||||
} from "./source-product-workflow";
|
||||
@ -88,6 +91,12 @@ export interface RegisterSourceHandlersOptions {
|
||||
"listLatestSyncCompletions" | "listLatestSyncRuns" | "listSyncPolicies"
|
||||
>
|
||||
| undefined;
|
||||
readonly sourceSyncPolicyVersionBinder?:
|
||||
| Pick<SourceProductWorkflowRepository, "rebindSyncPolicySourceVersion">
|
||||
| undefined;
|
||||
readonly sourceProductWorkflowService?:
|
||||
| Pick<SourceProductWorkflowService, "createSync">
|
||||
| undefined;
|
||||
readonly sources: SourceRepository;
|
||||
readonly spaces: KnowledgeSpaceRepository;
|
||||
readonly websiteCrawlConnector?: WebsiteCrawlConnector | undefined;
|
||||
@ -105,6 +114,8 @@ export function registerSourceHandlers({
|
||||
sourceCredentials,
|
||||
sourceDocumentMaterializer,
|
||||
sourceProductWorkflows,
|
||||
sourceProductWorkflowService,
|
||||
sourceSyncPolicyVersionBinder,
|
||||
sources,
|
||||
spaces,
|
||||
websiteCrawlConnector,
|
||||
@ -340,6 +351,23 @@ export function registerSourceHandlers({
|
||||
return context.json({ error: "Provider parameters cannot contain credentials" }, 400);
|
||||
}
|
||||
|
||||
const immutableMetadataKeys = [
|
||||
"datasource",
|
||||
"pluginId",
|
||||
"provider",
|
||||
"providerId",
|
||||
"providerKind",
|
||||
] as const;
|
||||
const requestsConfigurationSync =
|
||||
body.syncAfterUpdate === true &&
|
||||
(body.providerParameters !== undefined ||
|
||||
body.uri !== undefined ||
|
||||
(body.metadata !== undefined &&
|
||||
Object.keys(body.metadata).some((key) => key !== "syncPolicy")));
|
||||
if (requestsConfigurationSync && !sourceProductWorkflowService) {
|
||||
return context.json({ error: "Source workflow service is unavailable" }, 503);
|
||||
}
|
||||
|
||||
if (body.metadata?.syncPolicy !== undefined) {
|
||||
try {
|
||||
parseSourceSyncPolicy(body.metadata.syncPolicy);
|
||||
@ -354,6 +382,8 @@ export function registerSourceHandlers({
|
||||
|
||||
try {
|
||||
let source = null;
|
||||
let configurationChanged = false;
|
||||
let previousSourceVersion: number | undefined;
|
||||
const maxAttempts = body.expectedVersion === undefined ? 3 : 1;
|
||||
|
||||
for (let attempt = 0; attempt < maxAttempts; attempt += 1) {
|
||||
@ -366,6 +396,16 @@ export function registerSourceHandlers({
|
||||
break;
|
||||
}
|
||||
|
||||
for (const key of immutableMetadataKeys) {
|
||||
if (
|
||||
body.metadata &&
|
||||
Object.prototype.hasOwnProperty.call(body.metadata, key) &&
|
||||
body.metadata[key] !== fresh.metadata[key]
|
||||
) {
|
||||
return context.json({ error: `Source ${key} cannot be changed` }, 400);
|
||||
}
|
||||
}
|
||||
|
||||
if (body.expectedVersion !== undefined && body.expectedVersion !== fresh.version) {
|
||||
throw new SourceVersionConflictError(params.sourceId, body.expectedVersion);
|
||||
}
|
||||
@ -416,6 +456,15 @@ export function registerSourceHandlers({
|
||||
}
|
||||
}
|
||||
}
|
||||
configurationChanged =
|
||||
(uri !== undefined && uri !== fresh.uri) ||
|
||||
(body.providerParameters !== undefined &&
|
||||
!isDeepStrictEqual(metadata?.parameters, fresh.metadata.parameters)) ||
|
||||
(body.metadata !== undefined &&
|
||||
Object.entries(body.metadata).some(
|
||||
([key]) =>
|
||||
key !== "syncPolicy" && !isDeepStrictEqual(metadata?.[key], fresh.metadata[key]),
|
||||
));
|
||||
source = await sources.update({
|
||||
expectedVersion: fresh.version,
|
||||
id: params.sourceId,
|
||||
@ -433,6 +482,7 @@ export function registerSourceHandlers({
|
||||
...(body.status === undefined ? {} : { status: body.status }),
|
||||
...(uri === undefined ? {} : { uri }),
|
||||
});
|
||||
if (source) previousSourceVersion = fresh.version;
|
||||
break;
|
||||
} catch (error) {
|
||||
if (
|
||||
@ -450,6 +500,29 @@ export function registerSourceHandlers({
|
||||
if (!source) {
|
||||
return context.json({ error: "Source not found" }, 404);
|
||||
}
|
||||
if (previousSourceVersion !== undefined && sourceSyncPolicyVersionBinder) {
|
||||
await sourceSyncPolicyVersionBinder.rebindSyncPolicySourceVersion({
|
||||
expectedSourceVersion: previousSourceVersion,
|
||||
knowledgeSpaceId: params.id,
|
||||
sourceId: source.id,
|
||||
sourceVersion: source.version,
|
||||
tenantId: subject.tenantId,
|
||||
});
|
||||
}
|
||||
|
||||
if (body.syncAfterUpdate && configurationChanged && source.status !== "disabled") {
|
||||
const workflow = await sourceProductWorkflowService?.createSync({
|
||||
...sourceWorkflowPrincipal(context),
|
||||
idempotencyKey: `source-config-update:${source.id}:${source.version}`,
|
||||
knowledgeSpaceId: params.id,
|
||||
sourceId: source.id,
|
||||
});
|
||||
if (!workflow) throw new Error("Source workflow service became unavailable");
|
||||
return context.json(
|
||||
{ ...toSourceResponse(source), syncWorkflow: toPublicSourceWorkflowRun(workflow) },
|
||||
200,
|
||||
);
|
||||
}
|
||||
|
||||
return context.json(toSourceResponse(source), 200);
|
||||
} catch (error) {
|
||||
@ -1236,6 +1309,24 @@ export function registerSourceHandlers({
|
||||
});
|
||||
}
|
||||
|
||||
function sourceWorkflowPrincipal(context: Pick<LooseOpenApiContext, "get">) {
|
||||
const apiKey = context.get("authenticatedApiKey");
|
||||
const capabilityGrant = context.get("capabilityV2Grant");
|
||||
return {
|
||||
...(apiKey ? { apiKey } : {}),
|
||||
...(capabilityGrant
|
||||
? {
|
||||
capability: {
|
||||
contentScopeIds: capabilityGrant.contentScopeIds,
|
||||
grantId: capabilityGrant.grantId,
|
||||
},
|
||||
}
|
||||
: {}),
|
||||
callerKind: context.get("callerKind") ?? "interactive",
|
||||
subject: context.get("subject"),
|
||||
} as const;
|
||||
}
|
||||
|
||||
function sourceStatusWithSyncWorkflow(source: Source, run?: SourceWorkflowRun): Source["status"] {
|
||||
if (source.status === "disabled" || !run) return source.status;
|
||||
if (run.state === "failed" || run.state === "canceled") return "error";
|
||||
|
||||
@ -2184,6 +2184,36 @@ describe("database source-product workflow repository edge coverage", () => {
|
||||
expect(insert?.params[13]).toBeNull();
|
||||
});
|
||||
|
||||
it("rebinds a sync policy source version without changing its schedule or revision", async () => {
|
||||
const calls: DatabaseExecuteInput[] = [];
|
||||
const repository = createDatabaseSourceProductWorkflowRepository({
|
||||
database: testDatabase("postgres", async (input) => {
|
||||
calls.push(input);
|
||||
if (input.tableName === "source_sync_policies" && input.operation === "select") {
|
||||
return { rows: [syncPolicyRow()], rowsAffected: 1 };
|
||||
}
|
||||
return { rows: [], rowsAffected: input.operation === "select" ? 0 : 1 };
|
||||
}),
|
||||
});
|
||||
|
||||
await expect(
|
||||
repository.rebindSyncPolicySourceVersion({
|
||||
expectedSourceVersion: 1,
|
||||
knowledgeSpaceId,
|
||||
sourceId,
|
||||
sourceVersion: 4,
|
||||
tenantId,
|
||||
}),
|
||||
).resolves.toEqual({ ...syncPolicy(), expectedSourceVersion: 4 });
|
||||
|
||||
const update = calls.find(
|
||||
(call) => call.tableName === "source_sync_policies" && call.operation === "update",
|
||||
);
|
||||
expect(update?.params).toEqual([4, "sync-policy-a", 1]);
|
||||
expect(update?.sql).not.toContain('"next_run_at" =');
|
||||
expect(update?.sql).not.toContain('"revision" =');
|
||||
});
|
||||
|
||||
it("disables invalid due policies and handles scheduler races", async () => {
|
||||
await expect(
|
||||
createDatabaseSourceProductWorkflowRepository({
|
||||
@ -2191,11 +2221,17 @@ describe("database source-product workflow repository edge coverage", () => {
|
||||
}).enqueueDueSyncRuns({ limit: 1, maxExecutionAttempts: 2, now }),
|
||||
).resolves.toEqual([]);
|
||||
|
||||
const disabledSourceCalls: DatabaseExecuteInput[] = [];
|
||||
await expect(
|
||||
createDatabaseSourceProductWorkflowRepository({
|
||||
database: duePolicyDatabase({ sourceStatus: "disabled" }),
|
||||
database: duePolicyDatabase({ calls: disabledSourceCalls, sourceStatus: "disabled" }),
|
||||
}).enqueueDueSyncRuns({ limit: 1, maxExecutionAttempts: 2, now }),
|
||||
).resolves.toEqual([]);
|
||||
expect(
|
||||
disabledSourceCalls.some(
|
||||
(call) => call.tableName === "source_sync_policies" && call.operation === "update",
|
||||
),
|
||||
).toBe(false);
|
||||
|
||||
await expect(
|
||||
createDatabaseSourceProductWorkflowRepository({
|
||||
@ -2421,6 +2457,7 @@ function bulkMutationDatabase(
|
||||
|
||||
function duePolicyDatabase(
|
||||
options: {
|
||||
readonly calls?: DatabaseExecuteInput[] | undefined;
|
||||
readonly currentPolicy?: DatabaseRow | undefined;
|
||||
readonly missingSource?: boolean | undefined;
|
||||
readonly missingSpace?: boolean | undefined;
|
||||
@ -2431,6 +2468,7 @@ function duePolicyDatabase(
|
||||
): DatabaseAdapter {
|
||||
const candidate = syncPolicyRow();
|
||||
return testDatabase(dialect, async (input) => {
|
||||
options.calls?.push(input);
|
||||
if (input.tableName === "source_sync_policies") {
|
||||
if (input.operation === "update") {
|
||||
return { rows: [], rowsAffected: options.policyUpdateRowsAffected ?? 1 };
|
||||
|
||||
@ -1118,6 +1118,26 @@ export function createDatabaseSourceProductWorkflowRepository(input: {
|
||||
if (result.rowsAffected !== 1) policyConflict();
|
||||
return policy;
|
||||
}),
|
||||
rebindSyncPolicySourceVersion: (input) =>
|
||||
database.transaction(async (tx) => {
|
||||
const policy = await getPolicy(database, tx, input, true);
|
||||
if (!policy) return null;
|
||||
if (policy.expectedSourceVersion !== input.expectedSourceVersion) {
|
||||
throw new SourceWorkflowError(
|
||||
"SOURCE_SYNC_POLICY_SOURCE_CONFLICT",
|
||||
"Source sync policy changed concurrently",
|
||||
);
|
||||
}
|
||||
const updated = await tx.execute({
|
||||
maxRows: 0,
|
||||
operation: "update",
|
||||
params: [input.sourceVersion, policy.id, input.expectedSourceVersion],
|
||||
sql: `UPDATE ${q(database, policyTable)} SET ${q(database, "expected_source_version")} = ${p(database, 1)} WHERE ${q(database, "id")} = ${p(database, 2)} AND ${q(database, "expected_source_version")} = ${p(database, 3)};`,
|
||||
tableName: policyTable,
|
||||
});
|
||||
if (updated.rowsAffected !== 1) policyConflict();
|
||||
return { ...policy, expectedSourceVersion: input.sourceVersion };
|
||||
}),
|
||||
listDueSyncPolicies: async ({ cursor, limit, now }) => {
|
||||
listLimit(limit);
|
||||
const readLimit = limit + 1;
|
||||
@ -1273,14 +1293,14 @@ export function createDatabaseSourceProductWorkflowRepository(input: {
|
||||
policy.nextRunAt > now
|
||||
)
|
||||
continue;
|
||||
if (
|
||||
!sourceRow ||
|
||||
numberColumn(sourceRow, "version") !== policy.expectedSourceVersion ||
|
||||
stringColumn(sourceRow, "status") === "disabled"
|
||||
) {
|
||||
if (!sourceRow || numberColumn(sourceRow, "version") !== policy.expectedSourceVersion) {
|
||||
await disableSyncPolicy(database, tx, policy, now);
|
||||
continue;
|
||||
}
|
||||
// Disabling a Source pauses scheduling; it does not destroy the user's durable policy.
|
||||
// Keep nextRunAt unchanged so an overdue policy is admitted on the first scheduler pass
|
||||
// after the Source is enabled again.
|
||||
if (stringColumn(sourceRow, "status") === "disabled") continue;
|
||||
const scheduledFor = policy.nextRunAt;
|
||||
const run: SourceWorkflowRun = {
|
||||
...(isCapabilitySyncPolicy(policy)
|
||||
|
||||
@ -1039,16 +1039,33 @@ describe("in-memory source product workflow repository", () => {
|
||||
updatedAt: "2026-03-01T00:01:00.000Z",
|
||||
});
|
||||
expect(revised.revision).toBe(2);
|
||||
const rebound = await repository.rebindSyncPolicySourceVersion({
|
||||
expectedSourceVersion: 1,
|
||||
knowledgeSpaceId,
|
||||
sourceId: provider.sourceId,
|
||||
sourceVersion: 4,
|
||||
tenantId,
|
||||
});
|
||||
expect(rebound).toEqual({ ...revised, expectedSourceVersion: 4 });
|
||||
await expect(
|
||||
repository.rebindSyncPolicySourceVersion({
|
||||
expectedSourceVersion: 1,
|
||||
knowledgeSpaceId,
|
||||
sourceId: provider.sourceId,
|
||||
sourceVersion: 5,
|
||||
tenantId,
|
||||
}),
|
||||
).rejects.toMatchObject({ code: "SOURCE_SYNC_POLICY_SOURCE_CONFLICT" });
|
||||
await expect(
|
||||
repository.getSyncPolicy({ knowledgeSpaceId, sourceId: provider.sourceId, tenantId }),
|
||||
).resolves.toEqual(revised);
|
||||
).resolves.toEqual(rebound);
|
||||
await expect(
|
||||
repository.listSyncPolicies({
|
||||
knowledgeSpaceId,
|
||||
sourceIds: ["missing", provider.sourceId, provider.sourceId],
|
||||
tenantId,
|
||||
}),
|
||||
).resolves.toEqual([revised]);
|
||||
).resolves.toEqual([rebound]);
|
||||
await expect(
|
||||
repository.listSyncPolicies({
|
||||
knowledgeSpaceId,
|
||||
|
||||
@ -646,6 +646,26 @@ export function createInMemorySourceProductWorkflowRepository(input?: {
|
||||
policies.set(key, clonePolicy(policy));
|
||||
return clonePolicy(policy);
|
||||
},
|
||||
rebindSyncPolicySourceVersion: async ({
|
||||
expectedSourceVersion,
|
||||
knowledgeSpaceId,
|
||||
sourceId,
|
||||
sourceVersion,
|
||||
tenantId,
|
||||
}) => {
|
||||
const key = `${tenantId}\0${knowledgeSpaceId}\0${sourceId}`;
|
||||
const policy = policies.get(key);
|
||||
if (!policy) return null;
|
||||
if (policy.expectedSourceVersion !== expectedSourceVersion) {
|
||||
throw new SourceWorkflowError(
|
||||
"SOURCE_SYNC_POLICY_SOURCE_CONFLICT",
|
||||
"Source sync policy changed concurrently",
|
||||
);
|
||||
}
|
||||
const rebound = { ...policy, expectedSourceVersion: sourceVersion };
|
||||
policies.set(key, rebound);
|
||||
return clonePolicy(rebound);
|
||||
},
|
||||
getSyncPolicy: async ({ knowledgeSpaceId, sourceId, tenantId }) => {
|
||||
const policy = policies.get(`${tenantId}\0${knowledgeSpaceId}\0${sourceId}`);
|
||||
return policy ? clonePolicy(policy) : null;
|
||||
|
||||
@ -3089,6 +3089,13 @@ async function executeBulkAction(
|
||||
}),
|
||||
);
|
||||
if (!updated) throw runtimeError("SOURCE_NOT_FOUND", "Source not found");
|
||||
await input.repository.rebindSyncPolicySourceVersion({
|
||||
expectedSourceVersion: source.version,
|
||||
knowledgeSpaceId: updated.knowledgeSpaceId,
|
||||
sourceId: updated.id,
|
||||
sourceVersion: updated.version,
|
||||
tenantId: run.tenantId,
|
||||
});
|
||||
return;
|
||||
}
|
||||
throw runtimeError(
|
||||
|
||||
@ -391,6 +391,14 @@ export interface SourceProductWorkflowRepository {
|
||||
readonly run: NewSourceWorkflowRun;
|
||||
}): Promise<SourceWorkflowRun>;
|
||||
upsertSyncPolicy(input: SourceSyncPolicyRecord): Promise<SourceSyncPolicyRecord>;
|
||||
/** Rebinds an unchanged policy to a successful Source CAS without changing its schedule. */
|
||||
rebindSyncPolicySourceVersion(input: {
|
||||
readonly expectedSourceVersion: number;
|
||||
readonly knowledgeSpaceId: string;
|
||||
readonly sourceId: string;
|
||||
readonly sourceVersion: number;
|
||||
readonly tenantId: string;
|
||||
}): Promise<SourceSyncPolicyRecord | null>;
|
||||
listDueSyncPolicies(input: {
|
||||
readonly cursor?: string | undefined;
|
||||
readonly limit: number;
|
||||
|
||||
@ -69,6 +69,12 @@ export const UpdateSourceSchema = z
|
||||
return keys.length <= 50 && keys.every((key) => key.length >= 1 && key.length <= 255);
|
||||
}, "Provider parameters must contain at most 50 bounded keys")
|
||||
.optional(),
|
||||
/**
|
||||
* Product gateways set this when a successful configuration update must enqueue a durable
|
||||
* Source sync. Internal reconciliation writes leave it unset so metadata-only lifecycle
|
||||
* updates do not recursively create sync work.
|
||||
*/
|
||||
syncAfterUpdate: z.boolean().optional(),
|
||||
status: z.enum(["active", "syncing", "error", "disabled"]).optional(),
|
||||
uri: z.string().min(1).max(SOURCE_URI_MAX_LENGTH).optional(),
|
||||
})
|
||||
|
||||
@ -159,7 +159,7 @@ export const updateSourceRoute = createRoute({
|
||||
},
|
||||
responses: {
|
||||
200: {
|
||||
content: { "application/json": { schema: SourceResponseSchema } },
|
||||
content: { "application/json": { schema: SourceListItemResponseSchema } },
|
||||
description: "Updated source",
|
||||
},
|
||||
400: InvalidRequestResponse,
|
||||
@ -169,6 +169,10 @@ export const updateSourceRoute = createRoute({
|
||||
description:
|
||||
"Source was modified concurrently, or protected legacy credential-bearing parameters cannot be replaced",
|
||||
},
|
||||
503: {
|
||||
content: { "application/json": { schema: ErrorResponseSchema } },
|
||||
description: "Source workflow service unavailable",
|
||||
},
|
||||
401: UnauthorizedResponse,
|
||||
403: ForbiddenResponse,
|
||||
},
|
||||
|
||||
Loading…
Reference in New Issue
Block a user