[Async Batch Pipeline] Add AsyncAction support to actions-core#3768
Draft
sayan-das-in wants to merge 1 commit into
Draft
[Async Batch Pipeline] Add AsyncAction support to actions-core#3768sayan-das-in wants to merge 1 commit into
sayan-das-in wants to merge 1 commit into
Conversation
Contributor
There was a problem hiding this comment.
Pull request overview
Adds first-class async action support to the Actions Core (packages/core) destination-kit so cloud destinations can define and execute batched async workflows (with a follow-up poll step), and updates CLI tooling + tests accordingly.
Changes:
- Introduces
AsyncActionDefinitionplus async batch + poll execution paths (executeAsyncBatch,executeAsyncPoll) indestination-kit. - Exposes new async types from core entrypoints (
AsyncActionDefinition,AsyncPerformBatchResponse,PollPayload,PollResponse). - Updates CLI validation/type-generation to account for
asyncActions, and adds unit tests covering async batching + polling.
Reviewed changes
Copilot reviewed 8 out of 8 changed files in this pull request and generated 7 comments.
Show a summary per file
| File | Description |
|---|---|
| packages/core/src/index.ts | Re-exports new async action / polling types from core. |
| packages/core/src/destination-kit/types.ts | Minor formatting fix to Result interface comment/layout. |
| packages/core/src/destination-kit/index.ts | Adds asyncActions to destination definitions and adds executeAsyncBatch / executeAsyncPoll on Destination. |
| packages/core/src/destination-kit/action.ts | Defines async action types (AsyncActionDefinition, PollPayload, PollResponse) and implements AsyncAction runtime. |
| packages/core/src/tests/destination-kit.test.ts | Adds unit tests for destination initialization + async batch/poll behavior. |
| packages/core/src/tests/batching.test.ts | Adds unit tests focused on async batching/polling flows. |
| packages/cli/src/commands/validate.ts | Validates destinations by considering both actions and asyncActions. |
| packages/cli/src/commands/generate/types.ts | Generates field payload types for both sync and async actions. |
Codecov Report❌ Patch coverage is ❌ Your patch check has failed because the patch coverage (53.36%) is below the target coverage (80.00%). You can increase the patch coverage or adjust the target coverage. Additional details and impacted files@@ Coverage Diff @@
## main #3768 +/- ##
=========================================
Coverage 80.93% 80.93%
=========================================
Files 1348 1405 +57
Lines 25089 28265 +3176
Branches 5212 6120 +908
=========================================
+ Hits 20305 22876 +2571
- Misses 3837 4436 +599
- Partials 947 953 +6 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
sayan-das-in
force-pushed
the
async-batch-pipeline-core
branch
from
7b7872f to
ce920cb
Compare
sayan-das-in
force-pushed
the
async-batch-pipeline-core
branch
from
ce920cb to
6c30649
Compare
Comment on lines
+131
to
+135
| // Total number of events in the batch being processed. | ||
| totalEventsCount: number | ||
|
|
||
| // Number of events that was processed | ||
| validEventsCount: number |
Comment on lines
+1056
to
+1060
| const syncModeVal = this.definition.syncMode ? bundle.mapping?.['__segment_internal_sync_mode'] : undefined | ||
| const syncMode = isSyncMode(syncModeVal) ? syncModeVal : undefined | ||
| const matchingKey = bundle.mapping?.['__segment_internal_matching_key'] | ||
| const audienceMembership = bundle.data.map((d) => resolveAudienceMembership(d, syncMode)) | ||
|
|
Comment on lines
+785
to
+789
| let audienceSettings = {} as AudienceSettings | ||
| // All events should be batched on the same audience | ||
| if (events[0].context?.personas) { | ||
| audienceSettings = events[0].context?.personas?.audience_settings as AudienceSettings | ||
| } |
Comment on lines
+2557
to
+2560
| const pollPayload: PollPayload = { | ||
| jobId: 'test-job-123', | ||
| attempt: 1 | ||
| } |
Comment on lines
+2574
to
+2578
| stats: { | ||
| totalEventsCount: 100, | ||
| successfulEventsCount: 50, | ||
| failedEventsCount: 0 | ||
| } |
Comment on lines
+922
to
+924
| const destination = new Destination(asyncPollDestination) | ||
| const pollPayload: PollPayload = { jobId: 'poll-job-123', attempt: 1 } | ||
|
|
sayan-das-in
force-pushed
the
async-batch-pipeline-core
branch
from
6c30649 to
84ddbac
Compare
sayan-das-in
force-pushed
the
async-batch-pipeline-core
branch
from
84ddbac to
9bb16ce
Compare
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 6 out of 6 changed files in this pull request and generated 10 comments.
Comments suppressed due to low confidence (1)
packages/core/src/destination-kit/index.ts:862
executeDynamicFieldonly looks upactionSluginthis.actions. If async actions are expected to supportdynamicFields(theAsyncActionDefinitiontype includes them), dynamic field resolution for async actions will currently return[]. Consider checkingthis.asyncActions[actionSlug]as well and delegating toexecuteDynamicFieldonAsyncActionwhen present.
public async executeDynamicField(
actionSlug: string,
fieldKey: string,
data: ExecuteDynamicFieldInput<Settings, object>,
/**
* The dynamicFn argument is optional since it is only used by dynamic hook input fields. (For now)
*/
dynamicFn?: RequestFn<Settings, any, DynamicFieldResponse, AudienceSettings>
) {
const action = this.actions[actionSlug]
if (!action) {
return []
}
return action.executeDynamicField(fieldKey, data, dynamicFn)
}
| // The payload passed into a poll operation of an AsyncActionDefinition. | ||
| export type PollPayload = { | ||
| // Job ID returned from the performBatch response that identifies the batch request being polled | ||
| jobId: string |
Comment on lines
+144
to
+147
| granularResults?: MultiStatusResponse | ||
|
|
||
| // Batch results for destinations with all or nothing statuses | ||
| batchResults?: ActionDestinationSuccessResponse | ActionDestinationErrorResponse |
Comment on lines
+1094
to
+1095
| const { jobId, multiStatusResponse: batchMultiStatus } = performBatchResponse | ||
|
|
Comment on lines
+785
to
+789
| let audienceSettings = {} as AudienceSettings | ||
| // All events should be batched on the same audience | ||
| if (events[0].context?.personas) { | ||
| audienceSettings = events[0].context?.personas?.audience_settings as AudienceSettings | ||
| } |
Comment on lines
+824
to
+828
| throw new IntegrationError( | ||
| `Async action ${actionSlug} not found or does not support polling.`, | ||
| 'NotImplemented', | ||
| 501 | ||
| ) |
Comment on lines
+2555
to
+2561
| test('should throw error when async action is not found', async () => { | ||
| const destinationTest = new Destination(destinationWithAsyncAction) | ||
| const pollPayload: PollPayload = { | ||
| jobId: 'test-job-123', | ||
| attempt: 1 | ||
| } | ||
|
|
Comment on lines
+2571
to
+2579
| const pollResponse: PollResponse = { | ||
| jobId: 'test-job-123', | ||
| status: 'IN_PROGRESS', | ||
| stats: { | ||
| totalEventsCount: 100, | ||
| successfulEventsCount: 50, | ||
| failedEventsCount: 0 | ||
| } | ||
| } |
Comment on lines
+2647
to
+2652
| batchResults: { | ||
| status: 500, | ||
| errortype: 'UNKNOWN_ERROR', | ||
| errormessage: 'Destination API returned an error' | ||
| } | ||
| } |
Comment on lines
+922
to
+925
| const destination = new Destination(asyncPollDestination) | ||
| const pollPayload: PollPayload = { jobId: 'poll-job-123', attempt: 1 } | ||
|
|
||
| const result = await destination.executeAsyncPoll('asyncPollAction', { |
Comment on lines
+973
to
+979
| }, | ||
| batchResults: { | ||
| status: 500, | ||
| errortype: 'UNKNOWN_ERROR', | ||
| errormessage: 'External API failure' | ||
| } | ||
| } |
sayan-das-in
force-pushed
the
async-batch-pipeline-core
branch
from
9bb16ce to
54174d2
Compare
sayan-das-in
force-pushed
the
async-batch-pipeline-core
branch
from
54174d2 to
ef07ffd
Compare
| // The payload passed into a poll operation of an AsyncActionDefinition. | ||
| export type PollPayload = { | ||
| // Job ID returned from the performBatch response that identifies the batch request being polled | ||
| jobId: string |
| granularResults?: MultiStatusResponse | ||
|
|
||
| // Batch results for destinations with all or nothing statuses | ||
| batchResults?: ActionDestinationSuccessResponse | ActionDestinationErrorResponse |
Comment on lines
+2571
to
+2578
| const pollResponse: PollResponse = { | ||
| jobId: 'test-job-123', | ||
| status: 'IN_PROGRESS', | ||
| stats: { | ||
| totalEventsCount: 100, | ||
| successfulEventsCount: 50, | ||
| failedEventsCount: 0 | ||
| } |
Comment on lines
+2606
to
+2613
| const pollResponse: PollResponse = { | ||
| jobId: 'test-job-456', | ||
| status: 'COMPLETED', | ||
| stats: { | ||
| totalEventsCount: 2, | ||
| successfulEventsCount: 2, | ||
| failedEventsCount: 0 | ||
| }, |
Comment on lines
+2639
to
+2646
| const pollResponse: PollResponse = { | ||
| jobId: 'test-job-789', | ||
| status: 'FAILED', | ||
| stats: { | ||
| totalEventsCount: 10, | ||
| successfulEventsCount: 0, | ||
| failedEventsCount: 10 | ||
| }, |
Comment on lines
+2676
to
+2683
| const pollResponse: PollResponse = { | ||
| jobId: 'test-job-auth', | ||
| status: 'COMPLETED', | ||
| stats: { | ||
| totalEventsCount: 1, | ||
| successfulEventsCount: 1, | ||
| failedEventsCount: 0 | ||
| } |
sayan-das-in
force-pushed
the
async-batch-pipeline-core
branch
from
ef07ffd to
02f0e6e
Compare
sayan-das-in
force-pushed
the
async-batch-pipeline-core
branch
from
02f0e6e to
a3ff38d
Compare
| // The payload passed into a poll operation of an AsyncActionDefinition. | ||
| export type PollPayload = { | ||
| // Job ID returned from the performBatch response that identifies the batch request being polled | ||
| jobId: string |
Comment on lines
+132
to
+145
| // The response from a poll operation of an AsyncActionDefinition. | ||
| export type PollResponse = { | ||
| // Echo back the job ID being polled | ||
| jobId: string | ||
|
|
||
| // HTTP status code returned by the partner API during the poll operation | ||
| status: number | ||
|
|
||
| // The status of the poll operation | ||
| pollStatus: 'IN_PROGRESS' | 'COMPLETED' | 'FAILED' | 'ERROR' | ||
|
|
||
| // Return a multi-status response | ||
| // If not available, the status code will be used to determine the outcome | ||
| multiStatusResponse?: MultiStatusResponse |
Comment on lines
+2561
to
+2564
| const pollPayload: PollPayload = { | ||
| jobId: 'test-job-123', | ||
| attempt: 1 | ||
| } |
Comment on lines
+2575
to
+2583
| const pollResponse: PollResponse = { | ||
| jobId: 'test-job-123', | ||
| status: 'IN_PROGRESS', | ||
| stats: { | ||
| totalEventsCount: 100, | ||
| successfulEventsCount: 50, | ||
| failedEventsCount: 0 | ||
| } | ||
| } |
| multiStatusResponse.pushSuccessResponse({ status: 200, body: {}, sent: {} }) | ||
| multiStatusResponse.pushSuccessResponse({ status: 200, body: {}, sent: {} }) | ||
|
|
||
| mockPerformBatch.mockResolvedValue({ |
Comment on lines
+945
to
+947
| const destination = new Destination(asyncPollDestination) | ||
| const pollPayload: PollPayload = { jobId: 'poll-job-123', attempt: 1 } | ||
|
|
Comment on lines
+934
to
+941
| const pollResponse: PollResponse = { | ||
| jobId: 'poll-job-123', | ||
| status: 'IN_PROGRESS', | ||
| stats: { | ||
| successCount: 50, | ||
| failureCount: 0 | ||
| } | ||
| } |
sayan-das-in
force-pushed
the
async-batch-pipeline-core
branch
from
a3ff38d to
e3187d6
Compare
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
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.
This PR adds async actions support to Actions Core package.
These actions will be called from a different endpoint and doesn't any existing flow
Testing
Include any additional information about the testing you have completed to
ensure your changes behave as expected. For a speedy review, please check
any of the tasks you completed below during your testing.
Security Review
Please ensure sensitive data is properly protected in your integration.
type: 'password'New Destination Checklist
verioning-info.tsfile. example