Rate limits and retries

const API = "https://api.shiftlabs.org/v1";
const headers = {
  authorization: `Bearer ${process.env.SHIFTLABS_API_KEY}`,
  "content-type": "application/json",
};

// The same Idempotency-Key on every attempt: a retry never starts a second
// run.
async function startRun(workflow: string, payload: unknown, key: string) {
  for (let attempt = 1; attempt <= 5; attempt++) {
    const r = await fetch(`${API}/workflows/${workflow}/runs`, {
      method: "POST",
      headers: { ...headers, "idempotency-key": key },
      body: JSON.stringify(payload),
    }).catch(() => null);
    if (r?.ok) return (await r.json()) as { runId: string };
    if (r && r.status !== 429 && r.status !== 503) {
      throw new Error(`${r.status} ${await r.text()}`);
    }
    const wait = Number(r?.headers.get("retry-after") ?? 1);
    // A quota or the spending limit: an answer for a person, not a retry.
    if (wait > 60) throw new Error(`${r?.status} ${await r?.text()}`);
    // Clients sharing a budget spread out by adding up to a second.
    await new Promise((ok) => setTimeout(ok, (wait + Math.random()) * 1000));
  }
  throw new Error(`${workflow}: still refused after 5 attempts`);
}

const { runId } = await startRun("refund-check", { id: "ch_123" }, "ch_123");
console.log(runId);

import json
import os
import random
import time
import urllib.error
import urllib.request

API = "https://api.shiftlabs.org/v1"
KEY = os.environ["SHIFTLABS_API_KEY"]


def start_run(workflow: str, payload: object, key: str) -> dict:
    """The same Idempotency-Key on every attempt: a retry never starts a
    second run."""
    for _ in range(5):
        request = urllib.request.Request(
            f"{API}/workflows/{workflow}/runs",
            data=json.dumps(payload).encode(),
            headers={
                "authorization": f"Bearer {KEY}",
                "content-type": "application/json",
                "idempotency-key": key,
            },
        )
        try:
            with urllib.request.urlopen(request, timeout=30) as r:
                return json.load(r)
        except urllib.error.HTTPError as err:
            if err.code not in (429, 503):
                raise
            wait = int(err.headers.get("retry-after", "1"))
            # A quota or the spending limit: an answer for a person.
            if wait > 60:
                raise
        except (urllib.error.URLError, TimeoutError):
            wait = 1
        # Clients sharing a budget spread out by adding up to a second.
        time.sleep(wait + random.random())
    raise RuntimeError(f"{workflow}: still refused after 5 attempts")


print(start_run("refund-check", {"id": "ch_123"}, "ch_123")["runId"])

import { FatalError, workflow } from "shiftlabs";

export default workflow<{ email: string }>(
  async ({ payload, step, secrets }) => {
    return step.run(
      "add contact",
      { retries: { limit: 8, delay: "30 seconds", backoff: "exponential" } },
      async () => {
        const r = await fetch("https://api.example.com/contacts", {
          method: "POST",
          headers: {
            authorization: `Bearer ${secrets.CRM_KEY}`,
            "content-type": "application/json",
            "idempotency-key": step.idempotencyKey("add contact"),
          },
          body: JSON.stringify({ email: payload.email }),
        });
        // Worth another try after the step's delay.
        if (r.status === 429 || r.status >= 500) {
          throw new Error(`contacts answered ${r.status}`);
        }
        // Won't change on a retry.
        if (!r.ok) throw new FatalError(`contacts answered ${r.status}`);
        return r.json();
      },
    );
  },
);

import json

from workers import fetch
from shiftlabs import FatalError, workflow


@workflow
async def add_contact(ctx):
    async def attempt():
        r = await fetch(
            "https://api.example.com/contacts",
            method="POST",
            headers={
                "authorization": f"Bearer {ctx.secrets.CRM_KEY}",
                "content-type": "application/json",
                "idempotency-key": ctx.step.idempotency_key("add contact"),
            },
            body=json.dumps({"email": ctx.payload["email"]}),
        )
        # Worth another try after the step's delay.
        if r.status == 429 or r.status >= 500:
            raise Exception(f"contacts answered {r.status}")
        # Won't change on a retry.
        if not r.ok:
            raise FatalError(f"contacts answered {r.status}")
        return await r.json()

    return await ctx.step.run(
        "add contact",
        attempt,
        retries={"limit": 8, "delay": "30 seconds", "backoff": "exponential"},
    )

# Twice with the same key: both answer 202 with the same runId, and the
# second has the header Idempotent-Replayed: true.
for i in 1 2; do
  curl -si -X POST https://api.shiftlabs.org/v1/workflows/refund-check/runs \
    -H "authorization: Bearer $SHIFTLABS_API_KEY" \
    -H "content-type: application/json" \
    -H "idempotency-key: ch_123" \
    -d '{"id": "ch_123"}'
done

# One run, not two.
curl "https://api.shiftlabs.org/v1/runs?workflow=refund-check&limit=5" \
  -H "authorization: Bearer $SHIFTLABS_API_KEY"