Reconnect and recover gaps

import random
import time

import grpc
from polymarket.v1 import dropcopy_pb2, dropcopy_pb2_grpc  # generated from polymarket-protos.zip
from polymarket_auth import PolymarketClient                # canonical client: /connect#authentication

# The gRPC target is GRPC_TARGET in polymarket_auth.py — one place, both environments.
pmx = PolymarketClient()
PARTICIPANT_ID = "firms/20260821-examplefirminc-api-participant/participants/p-0f31c9"  # participantId from kyc.approved, verbatim
ACCOUNT        = "PASTE_PROVISIONED_ACCOUNT"  # provisionedAccount from kyc.approved, verbatim
RESUME_FILE    = "./dropcopy.resume"

BACKOFF_MIN   = 1.0    # seconds; also the floor set by the 1 open/sec rate on StreamRFQEvents
BACKOFF_MAX   = 30.0   # seconds; under the 10-minute ALB connection life
FATAL = {
    grpc.StatusCode.PERMISSION_DENIED,  # missing scope grant — re-mint after we add it
    grpc.StatusCode.ABORTED,            # suppressed ledger entry type requested
    grpc.StatusCode.INVALID_ARGUMENT,   # request defect
}


def subscribe_once() -> None:
    """One connection. Returns when the stream ends; raises on a fatal status."""
    with pmx.channel() as channel:
        stub = dropcopy_pb2_grpc.DropCopyAPIStub(channel)
        request = dropcopy_pb2.CreateDropCopySubscriptionRequest(accounts=[ACCOUNT])
        try:
            with open(RESUME_FILE, "rb") as fh:
                token = fh.read()
            if token:
                request.resume_token = token
        except FileNotFoundError:
            pass  # no token yet: fresh subscription, not a replay

        stream = stub.CreateDropCopySubscription(
            request,
            # Built per attempt, so each reconnect carries a live token.
            metadata=pmx.metadata(participant_id=PARTICIPANT_ID),
        )
        for message in stream:
            handle(message)  # must be idempotent: delivery is at-least-once
            if message.resume_token:
                with open(RESUME_FILE, "wb") as fh:
                    fh.write(message.resume_token)


def handle(message) -> None:
    print(message)


def run_forever() -> None:
    backoff = BACKOFF_MIN
    while True:
        try:
            subscribe_once()
            backoff = BACKOFF_MIN  # clean end: the ALB 10-minute close lands here
            print("stream closed; reconnecting immediately")
            time.sleep(BACKOFF_MIN)
            continue
        except grpc.RpcError as err:
            code, details = err.code(), err.details()
            if code in FATAL:
                raise RuntimeError(f"not retryable: {code.name}: {details}") from err
            # 13 INTERNAL "Subscription manager revoked session" is terminal for the session and
            # retryable as a NEW subscribe. SocketError / UNAVAILABLE is the 10-minute ALB close.
            print(f"reconnecting after {code.name}: {details}")

        time.sleep(backoff * random.random())  # full jitter
        backoff = min(backoff * 2, BACKOFF_MAX)


if __name__ == "__main__":
    run_forever()

import { readFileSync, writeFileSync } from "node:fs";
import * as grpc from "@grpc/grpc-js";
import * as protoLoader from "@grpc/proto-loader";
import { PolymarketClient } from "./polymarketAuth"; // canonical client: /connect#authentication

const pmx = new PolymarketClient();

const GRPC_TARGET = "grpc-api.preprod.polymarketexchange.com:443";
const PARTICIPANT_ID =
  "firms/20260821-examplefirminc-api-participant/participants/p-0f31c9"; // participantId from kyc.approved, verbatim
const ACCOUNT = "PASTE_PROVISIONED_ACCOUNT"; // provisionedAccount from kyc.approved, verbatim
const RESUME_FILE = "./dropcopy.resume";
const PROTO = "./protos/polymarket/v1/dropcopy.proto"; // from polymarket-protos.zip

const BACKOFF_MIN_MS = 1_000;  // also the floor set by the 1 open/sec rate on StreamRFQEvents
const BACKOFF_MAX_MS = 30_000; // under the 10-minute ALB connection life
const FATAL = new Set<number>([
  grpc.status.PERMISSION_DENIED, // missing scope grant — re-mint after we add it
  grpc.status.ABORTED,           // suppressed ledger entry type requested
  grpc.status.INVALID_ARGUMENT,  // request defect
]);

const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

function subscribeOnce(): Promise<void> {
  return new Promise<void>(async (resolve, reject) => {
    const pkg = grpc.loadPackageDefinition(
      protoLoader.loadSync(PROTO, { keepCase: true, defaults: true, includeDirs: ["./protos"] }),
    ) as any;
    const client = new pkg.polymarket.v1.DropCopyAPI(GRPC_TARGET, grpc.credentials.createSsl());

    // Built per attempt, so each reconnect carries a live token.
    const metadata = await pmx.metadata(PARTICIPANT_ID);

    const request: Record<string, unknown> = { accounts: [ACCOUNT] };
    try {
      const token = readFileSync(RESUME_FILE);
      if (token.length > 0) request.resume_token = token;
    } catch {
      /* no token yet: fresh subscription, not a replay */
    }

    const stream = client.CreateDropCopySubscription(request, metadata);
    stream.on("data", (message: any) => {
      handle(message); // must be idempotent: delivery is at-least-once
      if (message.resume_token?.length) writeFileSync(RESUME_FILE, message.resume_token);
    });
    stream.on("error", (err: grpc.ServiceError) => reject(err));
    stream.on("end", resolve);
  });
}

function handle(message: unknown): void {
  console.log(message);
}

export async function runForever(): Promise<never> {
  let backoff = BACKOFF_MIN_MS;
  for (;;) {
    try {
      await subscribeOnce();
      backoff = BACKOFF_MIN_MS; // clean end: the ALB 10-minute close lands here
      console.log("stream closed; reconnecting immediately");
      await sleep(BACKOFF_MIN_MS);
      continue;
    } catch (err) {
      const e = err as grpc.ServiceError;
      if (FATAL.has(e.code)) throw new Error(`not retryable: ${grpc.status[e.code]}: ${e.details}`);
      // 13 INTERNAL "Subscription manager revoked session" is terminal for the session and retryable
      // as a NEW subscribe. UNAVAILABLE / socket close is the 10-minute ALB close.
      console.log(`reconnecting after ${grpc.status[e.code]}: ${e.details}`);
    }
    await sleep(Math.random() * backoff); // full jitter
    backoff = Math.min(backoff * 2, BACKOFF_MAX_MS);
  }
}

runForever().catch((err) => {
  console.error(err);
  process.exit(1);
});

package main

import (
	"context"
	"fmt"
	"log"
	"math/rand"
	"os"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/status"

	pmauth "example.com/yourfirm/pmauth"        // canonical client: /connect#authentication
	dc "github.com/your-firm/gen/polymarket/v1" // generated from polymarket-protos.zip
)

const (
	grpcTarget    = "grpc-api.preprod.polymarketexchange.com:443"
	participantID = "firms/20260821-examplefirminc-api-participant/participants/p-0f31c9" // from kyc.approved, verbatim
	account       = "PASTE_PROVISIONED_ACCOUNT"                                            // provisionedAccount from kyc.approved, verbatim
	resumeFile    = "./dropcopy.resume"

	backoffMin = 1 * time.Second  // also the floor set by the 1 open/sec rate on StreamRFQEvents
	backoffMax = 30 * time.Second // under the 10-minute ALB connection life
)

// Missing scope grant, suppressed ledger entry type, request defect: none of these fix themselves.
var fatal = map[codes.Code]bool{
	codes.PermissionDenied: true,
	codes.Aborted:          true,
	codes.InvalidArgument:  true,
}

// One auth client for the life of the process: it caches the token and re-mints
// at expires_in minus 30 s.
var pm = pmauth.New()

func subscribeOnce(ctx context.Context) error {
	// Built per attempt, so each reconnect carries a live token.
	callMD, err := pm.Metadata(participantID)
	if err != nil {
		return err
	}

	conn, err := grpc.NewClient(grpcTarget, grpc.WithTransportCredentials(
		credentials.NewClientTLSFromCert(nil, ""),
	))
	if err != nil {
		return err
	}
	defer conn.Close()

	ctx = metadata.NewOutgoingContext(ctx, callMD)

	req := &dc.CreateDropCopySubscriptionRequest{Accounts: []string{account}}
	if resume, err := os.ReadFile(resumeFile); err == nil && len(resume) > 0 {
		req.ResumeToken = resume // no token yet means a fresh subscription, not a replay
	}

	stream, err := dc.NewDropCopyAPIClient(conn).CreateDropCopySubscription(ctx, req)
	if err != nil {
		return err
	}

	for {
		message, err := stream.Recv()
		if err != nil {
			return err
		}
		fmt.Println(message.String()) // handler must be idempotent: delivery is at-least-once
		if t := message.GetResumeToken(); len(t) > 0 {
			if err := os.WriteFile(resumeFile, t, 0o600); err != nil {
				return err
			}
		}
	}
}

func main() {
	backoff := backoffMin
	for {
		err := subscribeOnce(context.Background())
		code := status.Code(err)
		if fatal[code] {
			log.Fatalf("not retryable: %s: %s", code, status.Convert(err).Message())
		}
		// 13 INTERNAL "Subscription manager revoked session" is terminal for the session and
		// retryable as a NEW subscribe. Unavailable is the 10-minute ALB close.
		log.Printf("reconnecting after %s: %s", code, status.Convert(err).Message())

		time.Sleep(time.Duration(rand.Int63n(int64(backoff)))) // full jitter
		if backoff *= 2; backoff > backoffMax {
			backoff = backoffMax
		}
	}
}