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(
-    await fetch(
+    async () =>
-      `https://api.polygon.io/v3/reference/options/contracts?underlying_ticker=${symbol}&as_of=${date}&sort=ticker&limit=1000&apiKey=${await getApiKey()}`
+      (await (
-    )
+        await fetch(
-  ).json()) as PolygonOptionContractsResponse;
+          `https://api.polygon.io/v3/reference/options/contracts?underlying_ticker=${symbol}&as_of=${date}&sort=ticker&limit=1000&apiKey=${await getApiKey()}`
+        )
+      ).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(
-      await fetch(url)
+      async () =>
-    ).json()) as PolygonOptionContractAggregatesResponse;
+        (await (
+          await fetch(url)
+        ).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 clickhouse.insert({
+      await pRetry(
-        table: "option_contract_existences",
+        async () => {
-        values: batch,
+          await clickhouse.insert({
-        format: "JSONEachRow",
+            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 clickhouse.insert({
+        await pRetry(
-          table: "option_contract_aggregates",
+          async () => {
-          values: batch,
+            await clickhouse.insert({
-          format: "JSONEachRow",
+              table: "option_contract_aggregates",
-        });
+              values: batch,
+              format: "JSONEachRow",
+            });
+          },
+          { forever: true, factor: 2, maxTimeout: 120000 }
+        );
       }
     }
     await setPullOptionContractAggregatesState(ticker, {