Merge branch 'staging'

frak-id · Nov 20, 2024 · c0f7fcf · c0f7fcf
2 parents 1ebc3d7 + fe53994
commit c0f7fcf
Show file tree

Hide file tree

Showing 2 changed files with 46 additions and 20 deletions.
diff --git a/infra/components/ServiceTargets.ts b/infra/components/ServiceTargets.ts
@@ -105,6 +105,7 @@ export class ServiceTargets extends Component {
                                     timeout: toSeconds(health.timeout),
                                     matcher: health.successCodes,
                                 },
+                                deregistrationDelay: 20,
                             },
                             { parent: this }
                         );

diff --git a/packages/ponder/config/configBuilder.ts b/packages/ponder/config/configBuilder.ts
@@ -31,6 +31,16 @@ import {
     productRegistryAbi,
 } from "../abis/registryAbis";
 
+type GetLogsRpcType = Extract<
+    PublicRpcSchema[number],
+    { Method: "eth_getLogs" }
+>;
+
+type GetBlockByNumber = Extract<
+    PublicRpcSchema[number],
+    { Method: "eth_getBlockByNumber" }
+>;
+
 /**
  * Custom transport with some failsafe options specific for envio upstream
  * @param initialTransport
@@ -48,34 +58,49 @@ function safeClient(initialTransport: Transport): Transport {
                 if (
                     body.method === "eth_getLogs" &&
                     Array.isArray(body.params) &&
-                    body.params?.[0]?.blockHash
+                    (body.params as GetLogsRpcType["Parameters"])?.[0]
+                        ?.blockHash
                 ) {
-                    const requestedBlockHash = body.params?.[0]?.blockHash;
+                    const requestedBlockHash = (
+                        body.params as GetLogsRpcType["Parameters"]
+                    )?.[0]?.blockHash;
                     if (!requestedBlockHash) {
                         throw new Error("Missing blockHash parameter");
                     }
 
                     // Perform the request
-                    const response = (await transport.request(body)) as Extract<
-                        PublicRpcSchema[number],
-                        { Method: "eth_getLogs" }
-                    >["ReturnType"];
+                    const response = (await transport.request(
+                        body
+                    )) as GetLogsRpcType["ReturnType"];
 
                     // Filter out logs with a different blockHash
                     //  envio can leak parent / child block logs in the response
-                    return response.filter((log) => {
-                        if (log.blockHash !== requestedBlockHash) {
-                            console.log(
-                                "Filtering out log with different blockHash",
-                                {
-                                    requestedBlockHash,
-                                    logBlockHash: log.blockHash,
-                                }
-                            );
-                            return false;
-                        }
-                        return true;
-                    });
+                    const filteredResponse = response.filter(
+                        (log) => log.blockHash === requestedBlockHash
+                    );
+                    if (filteredResponse.length !== response.length) {
+                        console.log(
+                            `Filtered out ${
+                                response.length - filteredResponse.length
+                            } logs cause of mismatching blockHash, requested: ${requestedBlockHash}`
+                        );
+                    }
+                    return filteredResponse;
+                }
+
+                // If that's an eth_getBlockByNumber request, with latest block, do some manual thread lock, to ensure this block is well synchronised across different rpcs
+                if (
+                    body.method === "eth_getBlockByNumber" &&
+                    Array.isArray(body.params) &&
+                    (body.params as GetBlockByNumber["Parameters"])?.[0] ===
+                        "latest"
+                ) {
+                    // Perform the request
+                    const response = await transport.request(body);
+                    // Lock the thread for 3s
+                    await new Promise((resolve) => setTimeout(resolve, 3_000));
+                    // Return the response
+                    return response;
                 }
 
                 // Otherwise, simple request
@@ -150,7 +175,7 @@ export function createEnvConfig<NetworkKey extends string>({
             [networkKey]: {
                 chainId: network.chainId,
                 transport: getTransport(network.chainId),
-                pollingInterval: 5_000,
+                pollingInterval: 30_000,
             },
         },
         // contracts config