handle any socket hangups

server/src/lib/polygon.ts

@@ -38,11 +38,15 @@ export async function* makeGetOptionContractsIterator(
   symbol: string,
   date: string
 ) {
-  let latestBatchResponse = (await (
+  let latestBatchResponse = await pRetry(
+    async () =>
+      (await (
         await fetch(
           `https://api.polygon.io/v3/reference/options/contracts?underlying_ticker=${symbol}&as_of=${date}&sort=ticker&limit=1000&apiKey=${await getApiKey()}`
         )
-  ).json()) as PolygonOptionContractsResponse;
+      ).json()) as PolygonOptionContractsResponse,
+    { forever: true, factor: 2, maxTimeout: 120000 }
+  );
   yield latestBatchResponse.results.map((result) => ({
     asOfDate: date,
     symbol,
@@ -53,9 +57,15 @@ export async function* makeGetOptionContractsIterator(
 
   // as long as there's a `next_url`, call that:
   while (latestBatchResponse.hasOwnProperty("next_url")) {
-    latestBatchResponse = (await (
+    latestBatchResponse = await pRetry(
-      await fetch(`${latestBatchResponse.next_url}&apiKey=${await getApiKey()}`)
+      async () =>
-    ).json()) as PolygonOptionContractsResponse;
+        (await (
+          await fetch(
+            `${latestBatchResponse.next_url}&apiKey=${await getApiKey()}`
+          )
+        ).json()) as PolygonOptionContractsResponse,
+      { forever: true, factor: 2, maxTimeout: 120000 }
+    );
     yield latestBatchResponse.results?.map((result) => ({
       asOfDate: date,
       symbol,
@@ -106,9 +116,13 @@ export async function* makeGetOptionContractAggregatesIterator({
     const asOfDate = currentDateAsDateObject.toISOString().substring(0, 10);
     const url = `https://api.polygon.io/v2/aggs/ticker/${optionContractTicker}/range/1/minute/${asOfDate}/${asOfDate}?adjusted=false&sort=asc&limit=50000&apiKey=${await getApiKey()}`;
     let latestBatchResponse;
-    latestBatchResponse = (await (
+    latestBatchResponse = await pRetry(
+      async () =>
+        (await (
           await fetch(url)
-    ).json()) as PolygonOptionContractAggregatesResponse;
+        ).json()) as PolygonOptionContractAggregatesResponse,
+      { forever: true, factor: 2, maxTimeout: 120000 }
+    );
     if (latestBatchResponse.status.toLowerCase() === "ok") {
       yield latestBatchResponse.results?.map((result) => ({
         symbol,

server/src/lib/sync.ts

@@ -3,6 +3,7 @@ import sqlite3 from "sqlite3";
 import { open } from "sqlite";
 import { clickhouse, query } from "./clickhouse.js";
 import { OptionContract } from "./polygon.js";
+import pRetry from "p-retry";
 
 const sqliteDb = await open({
   filename: "/tmp/sync-state.db",
@@ -119,11 +120,16 @@ export async function pullOptionContracts(symbol: string, date: string) {
       date
     )) {
       console.log(batch.length);
+      await pRetry(
+        async () => {
           await clickhouse.insert({
             table: "option_contract_existences",
             values: batch,
             format: "JSONEachRow",
           });
+        },
+        { forever: true, factor: 2, maxTimeout: 120000 }
+      );
     }
     await setPullOptionContractsState(symbol, date, {
       status: OptionContractSyncStatus.COMPLETED,
@@ -157,11 +163,16 @@ export async function pullOptionContractAggregates(
           new Date(batch[0].tsStart * 1000),
           new Date(batch[batch.length - 1].tsStart * 1000)
         );
+        await pRetry(
+          async () => {
             await clickhouse.insert({
               table: "option_contract_aggregates",
               values: batch,
               format: "JSONEachRow",
             });
+          },
+          { forever: true, factor: 2, maxTimeout: 120000 }
+        );
       }
     }
     await setPullOptionContractAggregatesState(ticker, {