PipedreamHQ / PipedreamHQ/pipedream
[FEATURE] Enable concurrency control, de-duplication and ordering of event source executions
- Dominant language
- JavaScript
- Stars
- 11.7k
- Forks
- 5.8k
- Avg merge
- 3d 10h
- Merged PRs (30d)
- 102
Description
**Is your feature request related to a problem? Please describe.**
Depending on the use case of a particular event source, it makes sense to process incoming events (e.g. webhook calls, timers, etc.) either in parallel or sequentially. For example, Google Drive's push notifications sent to subscribers do not contain information about the event that triggered the notification and the responsibility of putting together such information is delegated to the subscribers themselves. This means that if multiple notifications arrive very closely to each other subscribers could potentially overlap and duplicate the work (as well as overwhelming the downstream API's by doing multiple calls in parallel). In such cases it makes sense to queue the incoming notifications and process them sequentially.
**Describe the solution you'd like**
Ideally, an event source would have a parameter that allows this kind of control (similar to the `dedupe` parameter). Something like this:
```js
const googleDrive = require("../google_drive.app");
module.exports = {
key: "google_drive-new-object",
name: "New Object",
props: {
googleDrive,
db: "$.service.db",
},
inboundConfig: {
// The max amount of executions that an event source could be running at any
// given time. Certain guarantees could be offered here, like ensuring that
// writes to `$.service.db` will be readable by subsequent executions.
concurrency: 1,
// Optional, but would allow the queueing of inbound events to be deduped
// and ordered. Taken from:
// https://developers.google.com/drive/api/v3/push#headers. In most cases a
// single field is enough to provide this deduping, but in some other cases
// it might be necessary to consider multiple fields and values to make the
// decision
dedupeConditions: [
"$.headers.['X-Goog-Message-Number']",
"$.headers.['X-Goog-Resource-State'] !== 'sync'",
],
},
};
```
**Do you have a workaround?**
There is currently no workaround available for event sources.
**Comparable features in other tools?**
[Temporal](https://temporal.io/) offers a mechanism to queue executions similar to what AWS Simple Workflow does, which is to use [task queues](https://docs.temporal.io/docs/go/task-queues) with deduplication logic (but it's managed internally by the framework in order to provide certain consistency guarantees). Workers that poll from these task queues can also be [configured with certain concurrency limits](https://pkg.go.dev/go.temporal.io/sdk@v1.8.0/internal#WorkerOptions).
**Additional context**
To add more context to the illustrative Google Drive example, this is what a push notification from Google Drive looks like:
```json
{
"method": "POST",
"path": "/",
"query": {},
"headers": {
"x-forwarded-for": "74.125.210.1",
"x-forwarded-proto": "https",
"x-forwarded-port": "443",
"host": "0ed2a33acbd5ebc0e90f18cf21a886e0.m.pipedream.net",
"x-amzn-trace-id": "Root=1-60dfad3b-66e85e337a2027bc1eb60d1e",
"content-length": "0",
"accept": "*/*",
"x-goog-channel-id": "89daf86c-4750-47a8-ba32-becb4d55305f",
"x-goog-channel-expiration": "Sat, 03 Jul 2021 01:19:27 GMT",
"x-goog-resource-state": "change",
"x-goog-message-number": "44154",
"x-goog-resource-id": "wuuGPhOI1mtwbT3YGLtr2WvMRXc",
"x-goog-resource-uri": "https://www.googleapis.com/drive/v3/changes?includeCorpusRemovals=false&includeItemsFromAllDrives=false&includeRemoved=true&includeTeamDriveItems=false&pageSize=100&pageToken=20243&restrictToMyDrive=false&spaces=drive&supportsAllDrives=false&supportsTeamDrives=false&alt=json",
"user-agent": "APIs-Google; (+https://developers.google.com/webmasters/APIs-Google.html)",
"accept-encoding": "gzip, deflate, br"
},
"bodyRaw": "",
"body": ""
}
```
Please note that the notification does not provide any context or information regarding the original event that caused the notification, other than the Google Drive in which the event occurred, and the type of event (`change` in this case).
When multiple events happen very closely in time, Google Drive might send many similar notifications with little difference between them in terms of information. Essentially, what would change from one notification to the other is:
- `x-amzn-trace-id`: undocumented field, probably used internally by Google for traceability
- `x-goog-message-number`: a field that allows subscribers to differentiate between different notifications
- `x-goog-resource-uri`: an API URL from which subscribers can obtain a list of events and where the originating event will be listed. **This is key, because event though the message number will be different between notifications, this URL will likely be the same for several notifications, hence causing concurrent access to the same Google Drive events if these notifications are processed in parallel**.
Contributor guide
Assessment
This issue has not been assessed yet.