Skip to content

Commit 6651064

Browse files
authored
Implement composite cache (#765)
* Create a base class for response cache * compare cache & composite transport * docs * Comments * Implement writeEntries * Prevent sharing transport instance * Hide composite implementation * Comments
1 parent d90df8d commit 6651064

19 files changed

Lines changed: 959 additions & 200 deletions

File tree

README.md

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@ yarn # Install yarn dependencies
4040
- [Subscription](./docs/components/transport-types/subscription-transport.md)
4141
- [Streaming](./docs/components/transport-types/streaming-transport.md)
4242
- [Custom](./docs/components/transport-types/custom-transport.md)
43+
- [Composite](./docs/components/transport-types/composite-transport.md)
4344
- Guides
4445
- [Porting a v2 EA to v3](./docs/guides/porting-a-v2-ea-to-v3.md)
4546
- [Creating a new v3 EA](./docs/guides/creating-a-new-v3-ea.md)
Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
# Composite transport
2+
3+
Composite transport is a **framework feature** for multi-route endpoints: every registered child transport runs in parallel for the same endpoint, and successful cache writes are merged so only the “freshest” value wins. You enable it on the endpoint; you **do not** import or `new CompositeTransport(...)` in adapter code—that class is constructed internally when the conditions below are met.
4+
5+
Typical uses:
6+
7+
- Pair a low-latency stream (for example WebSocket) with a REST fallback so the cache still updates if the stream lags or drops.
8+
- Run two data paths for the same feed and keep whichever provider reports a newer `providerIndicatedTimeUnixMs`.
9+
10+
## How to use it
11+
12+
1. Define the endpoint with **`transportRoutes`** (not a single `transport` field). Register **at least two** named child transports on a [`TransportRoutes`](../../../src/transports/index.ts) instance. Transport names must be lowercase letters only (see `TransportRoutes.register`).
13+
2. Set **`enableCompositeTransport: true`** on the same [`AdapterEndpoint`](../../../src/adapter/endpoint.ts) params.
14+
3. Turn the behavior on at runtime by setting adapter setting **`COMPOSITE_TRANSPORT`** to `true` (for example env `COMPOSITE_TRANSPORT=true`, or your adapter’s settings prefix).
15+
16+
If `enableCompositeTransport` is `true` but there are fewer than two routes, construction throws. If `enableCompositeTransport` is `true` but **`COMPOSITE_TRANSPORT`** is `false` (the default), the endpoint keeps **normal multi-transport routing** (`customRouter`, request `transport`, or `defaultTransport`) so operators can flip composite mode without redeploying.
17+
18+
When **both** flags are true, [`AdapterEndpoint.initialize`](../../../src/adapter/endpoint.ts) replaces the route map with a single internal route whose transport is a `CompositeTransport` built from your previous route entries. From then on the framework treats the endpoint as having one logical transport that fans out to all children.
19+
20+
## Example
21+
22+
```typescript
23+
import { AdapterEndpoint } from '@chainlink/external-adapter-framework/adapter'
24+
import { TransportRoutes } from '@chainlink/external-adapter-framework/transports'
25+
26+
// wsTransport and restTransport are normal transports you already defined
27+
export const endpoint = new AdapterEndpoint({
28+
name: 'example',
29+
inputParameters,
30+
enableCompositeTransport: true,
31+
transportRoutes: new TransportRoutes<EndpointTypes>()
32+
.register('ws', wsTransport)
33+
.register('rest', restTransport),
34+
})
35+
```
36+
37+
Deploy or configure with **`COMPOSITE_TRANSPORT=true`** when you want parallel execution and merged caching for that endpoint.
38+
39+
## How it works (internals)
40+
41+
The framework’s `CompositeTransport` (see [`composite.ts`](../../../src/transports/composite.ts)) wires each child with a [`CompareResponseCache`](../../../src/cache/response-cache/compare.ts) instead of the raw endpoint cache: reads go through to the real cache, while writes are accepted only when the pending payload is newer than both the last value seen for that key on that child path and the value already stored, using **`timestamps.providerIndicatedTimeUnixMs`** (missing timestamps are treated as `0`). **`registerRequest`** and **`backgroundExecute`** are invoked on **every** child in parallel. There is no `foregroundExecute` on the composite itself; behavior comes entirely from the children.
42+
43+
Child names are the keys you passed to `register`; each child’s `initialize` receives that string as its `transportName`.
44+
45+
## Notes
46+
47+
- **Timestamps** — Children should populate `providerIndicatedTimeUnixMs` when they have a meaningful provider clock; otherwise merge order may not match business intent.
48+
- **Concurrency** — Delivery order across children is not guaranteed; the merge rule is strictly “larger `providerIndicatedTimeUnixMs` wins.”
49+
- **TTL** — TTL behavior flows through the compare cache with the composite’s transport name; see `CompareResponseCache.writeTTL` if you depend on per-transport TTL semantics.
50+
- **Errors** — Children still own parsing and errors; the composite only arbitrates successful cache updates between children.

docs/components/transports.md

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ The v3 framework provides transports to fetch data from a Provider using the com
1313
- [HTTP Transport](./transport-types/http-transport.md)
1414
- [Websocket Transport](./transport-types/websocket-transport.md)
1515
- [SSE Transport](./transport-types/sse-transport.md)
16+
- [Composite Transport](./transport-types/composite-transport.md)
1617
- [Custom Transport](./transport-types/custom-transport.md)
1718

1819
### Abstract Transports

docs/reference-tables/ea-settings.md

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
| CACHE_REDIS_URL | string | undefined | The URL of the Redis server. Format: [redis[s]:]//[[user][:password@]][host][:port][/db-number]?db=db-number[&password=bar[&option=value]]] | - Value must be a valid URL | |
2828
| CACHE_TYPE | enum | local | The type of cache to use throughout the EA | | |
2929
| CENSOR_SENSITIVE_LOGS | boolean | false | Controls whether the logging of sensitive information is enabled or disabled | | |
30+
| COMPOSITE_TRANSPORT | boolean | false | Whether to use enableCompositeTransport parameter in AdapterEndpoint | | |
3031
| CORRELATION_ID_ENABLED | boolean | true | Flag to enable correlation IDs for sent requests in logging | | |
3132
| DEBUG | boolean | false | Toggles debug mode | | |
3233
| DEBUG_ENDPOINTS | boolean | false | Whether to enable debug enpoints (/debug/\*) for this adapter. Enabling them might consume more resources. | | |

src/adapter/endpoint.ts

Lines changed: 25 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
1-
import { ResponseCache } from '../cache/response'
1+
import { SimpleResponseCache } from '../cache/response'
22
import { AdapterSettings } from '../config'
33
import { TransportRoutes } from '../transports'
4+
import { CompositeTransport } from '../transports/composite'
45
import {
56
AdapterRequest,
67
AdapterRequestData,
@@ -46,6 +47,7 @@ export class AdapterEndpoint<T extends EndpointGenerics> implements AdapterEndpo
4647
settings: T['Settings'],
4748
) => string
4849
defaultTransport?: string
50+
enableCompositeTransport?: boolean
4951

5052
constructor(params: AdapterEndpointParams<T>) {
5153
this.name = params.name
@@ -55,6 +57,13 @@ export class AdapterEndpoint<T extends EndpointGenerics> implements AdapterEndpo
5557
this.transportRoutes = params.transportRoutes
5658
this.customRouter = params.customRouter
5759
this.defaultTransport = params.defaultTransport
60+
this.enableCompositeTransport = params.enableCompositeTransport
61+
if (params.enableCompositeTransport && this.transportRoutes.routeNames().length < 2) {
62+
throw new AdapterError({
63+
statusCode: 400,
64+
message: `Composite transport requires at least 2 transports`,
65+
})
66+
}
5867
} else {
5968
this.transportRoutes = new TransportRoutes<T>().register(
6069
DEFAULT_TRANSPORT_NAME,
@@ -83,7 +92,7 @@ export class AdapterEndpoint<T extends EndpointGenerics> implements AdapterEndpo
8392
adapterSettings: T['Settings'],
8493
): Promise<void> {
8594
this.adapterName = adapterName
86-
const responseCache = new ResponseCache({
95+
const responseCache = new SimpleResponseCache({
8796
dependencies,
8897
adapterSettings: adapterSettings as AdapterSettings,
8998
adapterName,
@@ -96,6 +105,20 @@ export class AdapterEndpoint<T extends EndpointGenerics> implements AdapterEndpo
96105
responseCache,
97106
}
98107

108+
if (this.enableCompositeTransport && !adapterSettings.COMPOSITE_TRANSPORT) {
109+
logger.warn(
110+
`enableCompositeTransport true for endpoint "${this.name}", but COMPOSITE_TRANSPORT is not enabled`,
111+
)
112+
}
113+
114+
if (this.enableCompositeTransport && adapterSettings.COMPOSITE_TRANSPORT) {
115+
logger.debug(`Enabling composite transport for endpoint "${this.name}"...`)
116+
this.transportRoutes = new TransportRoutes<T>().register(
117+
DEFAULT_TRANSPORT_NAME,
118+
new CompositeTransport(Object.fromEntries(this.transportRoutes.entries())),
119+
)
120+
}
121+
99122
logger.debug(`Initializing transports for endpoint "${this.name}"...`)
100123
for (const [transportName, transport] of this.transportRoutes.entries()) {
101124
await transport.initialize(transportDependencies, adapterSettings, this.name, transportName)

src/adapter/types.ts

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -183,6 +183,9 @@ type MultiTransportAdapterEndpointParams<T extends EndpointGenerics> = {
183183

184184
/** If no value is returned from the custom router or the default (transport param), which transport to use */
185185
defaultTransport?: string
186+
187+
/** If true, roll all transportRoutes under a new CompositeTransport */
188+
enableCompositeTransport?: boolean
186189
}
187190

188191
/**

src/cache/response-cache/base.ts

Lines changed: 170 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,170 @@
1+
import { AdapterDependencies } from '../../adapter'
2+
import { AdapterSettings } from '../../config'
3+
import {
4+
AdapterResponse,
5+
makeLogger,
6+
ResponseGenerics,
7+
TimestampedAdapterResponse,
8+
TimestampedProviderResult,
9+
censor,
10+
censorLogs,
11+
TimestampedProviderErrorResponse,
12+
} from '../../util'
13+
import {
14+
InputParameters,
15+
InputParametersDefinition,
16+
TypeFromDefinition,
17+
} from '../../validation/input-params'
18+
import { Cache, calculateAdapterName, calculateCacheKey, calculateFeedId } from '../'
19+
import CensorList from '../../util/censor/censor-list'
20+
import { validator } from '../../validation/utils'
21+
22+
const logger = makeLogger('ResponseCache')
23+
24+
export abstract class ResponseCache<
25+
T extends { Parameters: InputParametersDefinition; Response: ResponseGenerics },
26+
> {
27+
cache: Cache<AdapterResponse<T['Response']>>
28+
inputParameters: InputParameters<T['Parameters']>
29+
adapterName: string
30+
endpointName: string
31+
adapterSettings: AdapterSettings
32+
dependencies: AdapterDependencies
33+
34+
constructor({
35+
inputParameters,
36+
adapterName,
37+
endpointName,
38+
adapterSettings,
39+
dependencies,
40+
}: {
41+
dependencies: AdapterDependencies
42+
adapterSettings: AdapterSettings
43+
adapterName: string
44+
endpointName: string
45+
inputParameters: InputParameters<T['Parameters']>
46+
}) {
47+
this.dependencies = dependencies
48+
this.cache = dependencies.cache as Cache<AdapterResponse<T['Response']>>
49+
this.inputParameters = inputParameters
50+
this.adapterName = adapterName
51+
this.endpointName = endpointName
52+
this.adapterSettings = adapterSettings
53+
}
54+
55+
/**
56+
* Sets responses in the adapter cache (adding necessary metadata and defaults)
57+
*
58+
* @param transportName - transport name
59+
* @param results - the entries to write to the cache
60+
*/
61+
abstract write(transportName: string, results: TimestampedProviderResult<T>[]): Promise<void>
62+
63+
/**
64+
* Sets responses with metadata in the adapter cache
65+
*
66+
* @param entries - the entries to write to the cache
67+
*/
68+
abstract writeEntries(
69+
entries: {
70+
key: string
71+
value: AdapterResponse<T['Response']>
72+
}[],
73+
): Promise<void>
74+
75+
/**
76+
* Sets a new TTL value for already cached responses in the adapter cache
77+
*
78+
* @param transportName - transport name
79+
* @param params - set of parameters that uniquely relate to the response
80+
* @param ttl - a new time in milliseconds until the response expires
81+
*/
82+
async writeTTL(
83+
transportName: string,
84+
params: TypeFromDefinition<T['Parameters']>[],
85+
ttl: number,
86+
): Promise<void> {
87+
for (const param of params) {
88+
const key = this.getCacheKey(transportName, param)
89+
this.cache.setTTL(key, ttl)
90+
}
91+
}
92+
93+
async get(key: string) {
94+
return this.cache.get(key)
95+
}
96+
97+
protected generateCacheEntry(
98+
transportNameForMeta: string,
99+
transportNameForCache: string,
100+
r: TimestampedProviderResult<T>,
101+
) {
102+
const censorList = CensorList.getAll()
103+
const { data, result, errorMessage } = r.response
104+
if (!errorMessage && data === undefined) {
105+
logger.warn('The "data" property of the response is undefined.')
106+
} else if (!errorMessage && result === undefined) {
107+
logger.warn('The "result" property of the response is undefined.')
108+
}
109+
let censoredResponse
110+
if (!censorList.length) {
111+
censoredResponse = r.response
112+
} else {
113+
try {
114+
censoredResponse = censor(r.response, censorList, true) as TimestampedAdapterResponse<
115+
T['Response']
116+
>
117+
} catch (error) {
118+
censorLogs(() => logger.error(`Error censoring response: ${error}`))
119+
censoredResponse = {
120+
statusCode: 502,
121+
errorMessage: 'Response could not be censored due to an error',
122+
timestamps: r.response.timestamps,
123+
}
124+
}
125+
}
126+
127+
const response: AdapterResponse<T['Response']> = {
128+
...censoredResponse,
129+
statusCode: (censoredResponse as TimestampedProviderErrorResponse).statusCode || 200,
130+
}
131+
132+
if (this.adapterSettings.METRICS_ENABLED && this.adapterSettings.EXPERIMENTAL_METRICS_ENABLED) {
133+
response.meta = {
134+
adapterName: calculateAdapterName(this.adapterName, r.params),
135+
transportName: transportNameForMeta,
136+
metrics: {
137+
feedId: calculateFeedId(
138+
{
139+
adapterSettings: this.adapterSettings,
140+
},
141+
r.params,
142+
),
143+
},
144+
}
145+
}
146+
147+
if (response.timestamps?.providerIndicatedTimeUnixMs !== undefined) {
148+
const timestampValidator = validator.responseTimestamp()
149+
const error = timestampValidator.fn(response.timestamps?.providerIndicatedTimeUnixMs)
150+
if (error) {
151+
censorLogs(() => logger.warn(`Provider indicated time is invalid: ${error}`))
152+
}
153+
}
154+
155+
return {
156+
key: this.getCacheKey(transportNameForCache, r.params),
157+
value: response,
158+
} as const
159+
}
160+
161+
getCacheKey(transportName: string, params: TypeFromDefinition<T['Parameters']>) {
162+
return calculateCacheKey({
163+
transportName,
164+
data: params,
165+
adapterName: this.adapterName,
166+
endpointName: this.endpointName,
167+
adapterSettings: this.adapterSettings,
168+
})
169+
}
170+
}

0 commit comments

Comments
 (0)