feat(store-sync): apply logs from receipt #3215
Conversation
|
holic
changed the title
| return concat( | ||
| storageAdapterLock$.pipe( | ||
| first((lock) => lock === false), | ||
| tap(() => storageAdapterLock$.next(true)), | ||
| ignoreElements(), | ||
| ), | ||
| storedBlock$.pipe( | ||
| tap(({ blockNumber, logs }) => { | ||
| debug("stored", logs.length, "logs for block", blockNumber); | ||
| lastBlockNumberProcessed = blockNumber; | ||
| }), | ||
| ), | ||
| of(true).pipe( | ||
| concatMap(async () => { | ||
| if (lastBlockNumberProcessed != null) { | ||
| await applyOptimisticLogs(lastBlockNumberProcessed); | ||
| } | ||
| }), | ||
| tap(() => storageAdapterLock$.next(false)), | ||
| ignoreElements(), | ||
| ), | ||
| ); | ||
| }), |
There was a problem hiding this comment.
does the storageAdapterLock$ logic need to be rxjs based or could this be a boolean variable? i always need extra brain cycles to think through rxjs logic
There was a problem hiding this comment.
it's an observable because we need to subscribe to it in two places (one here to wait for the lock to free up and one in waitForTransaction)
There was a problem hiding this comment.
Maybe it's because it's late but i've stared at this part of the code for a while and am still not fully sure what's going on 🙈 What is the concat of three streams doing, out of which the first and last have a ignoreElements? why is the last stream of(true)?
There was a problem hiding this comment.
couldn't we do something like
for(const block of storedBlocks) {
storageAdapterLock.next(true);
await applyOptimisticLogs(block.blockNumber);
storageAdapterLock.next(false);
}or
storedBlock$.pipe(concatMap(block => {
storageAdapterLock$.next(true);
await applyOptimisticLogs(lastBlockNumberProcessed);
storageAdapterLock$.next(false);
}))?
There was a problem hiding this comment.
Okay this took a bunch of iteration to get here but here's the idea:
The current approach turns an async generator into a stream of blocks, one emission per yield of the generator. I originally had your suggested approach above, but found we were applying optimistic logs for each block rather than each range of blocks, doing a lot more work than is necessary, especially when hydrating from RPC.
Instead what this does is takes the block range stream (storedBlock$ but could rename for clarity) and adds an operation before and after the stream: before to wait for + take out a lock and after to apply optimistic logs and release the lock. Concatenating arbitrary things to the nice stream of storedBlocks$ means we'd have to find/filter them out later, but we can encapsulate that with ignoreElements() which basically empties the stream we added to the start/end and only completes/errors the stream.
So it works like this:
- create a block stream/observable for the range of blocks
- before we start evaluating that stream (and thus fetching+storing data), force the stream to wait for the lock to release in case there are existing pending operations
- once unlocked, take out a lock for the duration of the range
- process each block in the range
- once range is done, apply optimistic logs and release the lock before exiting the stream
Co-authored-by: alvarius <[email protected]>
No description provided.