Scheduler foundation: Singer-compliant schema, poll framework, and ar… - #98
Merged
Merged
Conversation
…chitecture spec - Add Singer-style sync_cursors schema keyed on (stitch_id, stream_name) instead of connection_id — prevents cursor collision between stitches sharing a source connection; state_document default includes bookmarks, versions, currently_syncing - Add dsProjectCode (nullable bigint) to organization table with CHECK constraint capping at JS MAX_SAFE_INTEGER to fail loudly on truncation - Add poll framework types to Piece interface: PollWindow, PollRecord, PollPage, StreamDescriptor (discriminated union enforcing replicationKey required on INCREMENTAL at compile time), ReplicationKeyType, poll(), describeStreams() - Add full DolphinScheduler + CursorManager architecture spec aligned with singer-python state.py conventions (currently_syncing, offset, versions at top level); covers DS tenant isolation, lifecycle, security, and edge cases - Update T029/T030/T046/T047/T048/T049 task descriptions to reflect corrected stitch_id key, DS_INTERNAL_SECRET auth, and Singer-compliant interfaces - Fix identity test fixture to include dsProjectCode: null Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Contributor
|
Warning Rate limit exceeded
⌛ How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: ASSERTIVE Plan: Pro Run ID: 📒 Files selected for processing (1)
📝 WalkthroughWalkthroughAdds stateful cursor-based polling: new Changes
Sequence DiagramsequenceDiagram
participant DS as DolphinScheduler
participant API as POST /internal/scheduler/execute-stitch
participant CM as CursorManagerService
participant Piece as Connector Piece
participant DB as Database (sync_cursors)
participant Redis as Redis Lock
DS->>API: Trigger stitch (stitch_id)
API->>API: Validate DS bearer secret
API->>Redis: Attempt lock for stitch_id
alt lock acquired
API->>DB: Load stateDocument for stitch_id
DB-->>API: stateDocument (bookmarks, versions)
API->>CM: calculateWindow(stream, stateDocument)
CM-->>API: PollWindow { lowerBound, upperBound, replicationKeyType }
loop paginate
API->>Piece: poll(credentials, streamName, window, nextPageCursor?)
Piece-->>API: PollPage { records[], nextPageCursor? }
API->>DB: Write intermediate checkpoint to sync_cursors
API->>Downstream: write records
end
API->>DB: Final checkpoint commit (stateDocument)
API->>DS: Return 200 { status: "SUCCESS" }
else lock contention
API->>DS: Return 200 { status: "SKIPPED" }
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 3✅ Passed checks (3 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 4
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
packages/database/src/schema/identity.ts (1)
258-276:⚠️ Potential issue | 🟠 MajorAdd a uniqueness guard for
dsProjectCodeto preserve tenant isolation.
dsProjectCodeis now a persisted org→DS mapping, but there is no unique constraint/index on this column. Two organizations can accidentally share one DS project code, which is a cross-tenant isolation risk.🔧 Proposed fix
(table) => [ check("organization_id_not_sentinel", sql`${table.id} <> '__NULL__'`), + uniqueIndex("organization_ds_project_code_unique_idx") + .on(table.dsProjectCode) + .where(sql`${table.dsProjectCode} IS NOT NULL`), check( "ds_project_code_safe_integer", sql`${table.dsProjectCode} IS NULL OR ${table.dsProjectCode} <= 9007199254740991`, ),🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/database/src/schema/identity.ts` around lines 258 - 276, Add a uniqueness constraint/index for dsProjectCode to prevent multiple organizations from sharing the same DolphinScheduler project code: modify the table definition that declares dsProjectCode (identifier: dsProjectCode on the table variable) to add a unique index (similar to organization_slug_unique_idx) that enforces uniqueness only for non-null dsProjectCode values (a partial unique index WHERE "dsProjectCode" IS NOT NULL) so NULLs remain allowed but any assigned project code is unique across organizations; ensure the new index has a clear name (e.g., organization_ds_project_code_unique_idx) and coexists with the existing checks (organization_id_not_sentinel and ds_project_code_safe_integer).
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@docs/architecture/scheduling/scheduling.md`:
- Around line 354-386: The calculateWindow function currently reads
catalog.replicationKeyType when bookmark?.replication_key_type is undefined,
which can be invalid for non-INCREMENTAL streams; update calculateWindow to
assert at runtime that the stream is INCREMENTAL (or that a replication key type
exists) before using catalog.replicationKeyType — e.g., check the stream's sync
mode or throw/log and bail if not INCREMENTAL — and add a short comment in
calculateWindow explaining this precondition; reference calculateWindow,
StreamDescriptor.replicationKeyType, and StreamBookmark to locate and fix the
code.
- Around line 172-173: The spec's table entry for the state_document default is
out of sync with the DB schema; update the table row for `state_document` to use
the actual default used in the schema (`{ bookmarks: {}, versions: {},
currently_syncing: null }`) so the spec matches the code in `stitches.ts`;
locate the `state_document` table row in scheduling.md and replace the existing
`{"bookmarks":{}}` default with the schema default object to keep documentation
consistent with the `stitches.ts` definition.
In `@packages/connectors/src/framework/piece.ts`:
- Around line 43-51: Add a JSDoc note to PollRecord.replicationKeyValue
explaining the coercion contract: implementers may return string or number, but
the system coerces numeric values to string for storage/comparison (e.g.,
PollWindow.lowerBound/upperBound are strings) and
CursorManagerService.trackHighWaterMark may call String(value) / Number(value)
when comparing or storing high-water marks; this informs piece authors they can
return either type and how it will be treated internally.
In `@packages/database/src/schema/identity.ts`:
- Around line 268-271: The constraint named "ds_project_code_safe_integer" only
enforces an upper bound on table.dsProjectCode; update the SQL expression used
in the check() call to enforce the full JS safe integer range by requiring the
column to be NULL or between -9007199254740991 and 9007199254740991 (or
equivalently add a >= -9007199254740991 check alongside the existing <= check)
so values below the negative safe limit are rejected; locate the check(...)
invocation for "ds_project_code_safe_integer" and modify its sql`${...}`
expression accordingly.
---
Outside diff comments:
In `@packages/database/src/schema/identity.ts`:
- Around line 258-276: Add a uniqueness constraint/index for dsProjectCode to
prevent multiple organizations from sharing the same DolphinScheduler project
code: modify the table definition that declares dsProjectCode (identifier:
dsProjectCode on the table variable) to add a unique index (similar to
organization_slug_unique_idx) that enforces uniqueness only for non-null
dsProjectCode values (a partial unique index WHERE "dsProjectCode" IS NOT NULL)
so NULLs remain allowed but any assigned project code is unique across
organizations; ensure the new index has a clear name (e.g.,
organization_ds_project_code_unique_idx) and coexists with the existing checks
(organization_id_not_sentinel and ds_project_code_safe_integer).
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: 8206bd63-b189-40b8-8ac0-3507d63642e0
📒 Files selected for processing (7)
apps/api/src/db/schema.tsdocs/architecture/master/tasks.mddocs/architecture/scheduling/scheduling.mdpackages/connectors/src/framework/piece.tspackages/database/src/schema/identity.tspackages/database/src/schema/stitches.tspackages/identity/src/adapters/drizzle-tenant.adapter.spec.ts
Contributor
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@docs/architecture/scheduling/scheduling.md`:
- Around line 167-176: The current docs show the sync_cursors table and its
state_document column holding multi-stream Singer state, which contradicts the
one-row-per-(stitch_id, stream_name) invariant; update the documentation for
sync_cursors (mentioning sync_cursors, state_document, stream_name, stitch_id)
to state explicitly that each row's state_document must contain only the
bookmark/version entry for that row's stream_name (not multiple streams),
replace the multi-stream JSON example with a single-stream JSON containing only
the corresponding stream entry, and make the same clarification/change in the
later section (the part covering lines 186–217) so the invariant is consistent
throughout.
In `@packages/connectors/src/framework/piece.ts`:
- Around line 67-72: The poll() contract is ambiguous because it receives only
credentials, window, and a string cursor which cannot identify which stream
(from describeStreams()) it targets nor preserve connector-specific
bookmark.offset shape; update the API so poll() accepts a structured cursor
including streamName and the connector bookmark (e.g., an object with streamName
and bookmark fields) and update PollPage to include streamName and a typed
nextPageCursor that can round-trip the bookmark; specifically modify the poll()
signature and any usages, adjust the PollPage interface (and related types
around lines ~135-156 and ~188-191) to carry streamName and a connector bookmark
payload instead of a plain string, and ensure scheduler state keys (stitchId,
streamName) align with the new cursor shape so the first page can
deterministically target a stream and subsequent pages can restore
bookmark.offset.
- Around line 29-40: Change PollWindow from a single interface into a
discriminated union keyed by replicationKeyType: define separate types (e.g.,
PollWindowTimestamp, PollWindowNumeric, PollWindowOpaque) each with
replicationKeyType set to the appropriate ReplicationKeyType literal and only
the bounds that make sense (Timestamp: lowerBound and upperBound as ISO strings;
Numeric: lowerBound and optional/typed numeric bound if applicable; Opaque:
lowerBound/token only, no mandatory upperBound). Replace the existing PollWindow
export with the union of those types and update any usages (including the poll()
signature and any code referencing upperBound/lowerBound) to handle the union
via the replicationKeyType discriminator.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: 88e5d8f0-b979-4e9d-a22c-d4a2e7c0def1
📒 Files selected for processing (3)
docs/architecture/scheduling/scheduling.mdpackages/connectors/src/framework/piece.tspackages/database/src/schema/identity.ts
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@docs/architecture/scheduling/scheduling.md`:
- Around line 563-592: The compose snippet uses simple depends_on entries
(ds-master, ds-worker, ds-api, ds-alert) which only wait for container start,
not readiness; update the snippet to reference health-based dependencies for the
backing services (postgres, zookeeper) and add corresponding healthcheck
definitions for those services so Docker Compose can use condition:
service_healthy; specifically, modify the depends_on blocks for ds-master
(postgres, zookeeper) and for ds-worker/ds-api to reference postgres (and
ds-master where applicable) with condition: service_healthy, and add healthcheck
configurations for postgres and zookeeper services to probe readiness (e.g.,
pg_isready/HTTP/TCP checks), ensuring service names ds-master, ds-worker,
ds-api, ds-alert, postgres, and zookeeper are updated accordingly.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: 17ed8d8a-473d-4d1c-be0d-86bfa5dda4ed
📒 Files selected for processing (2)
docs/architecture/scheduling/scheduling.mdpackages/connectors/src/framework/piece.ts
This was referenced Mar 23, 2026
Merged
Merged
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
…chitecture spec
Summary by CodeRabbit
New Features
Database
Documentation
Tests