From 6f78cc614e75b7ba4da7e01aa97a984478f3b46d Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 26 Jul 2026 17:12:23 +0000 Subject: [PATCH 1/2] Replace the three upload Workflows with inline retries and throttling Cloudflare are introducing per-step billing for Workflows. The weather station reports roughly every 30 seconds and each observation created three workflow instances (Windguru, Met Office, Windy), which is around 255k instances a month - expensive, and more machinery than the job needs. Each of those workflows was a single step.do() wrapping one fetch, purely to get durable retry, and the observation is already saved to D1 before the workflow is created. The uploads now run inline under ctx.waitUntil with their own retry loop: three attempts spread over ~18 seconds, five second timeout each. That is slightly less patient than the workflows were, but retries no longer outlive the invocation and a failed observation is superseded ~30 seconds later anyway. Each service is also throttled to its own minimum interval - two minutes for Windguru, five for Met Office and Windy - using a new upload_state table in D1. Windy rejects anything sent inside five minutes, so most of its uploads were previously created only to be discarded. This cuts outbound calls from ~8,500 a day to ~1,300 and removes nearly all of the rate limit rejections. upload_state also records the last success and last error per service, which replaces the visibility the Workflows dashboard used to give. D1 is used rather than KV for the throttle state because KV reads are cached for a minimum of 60 seconds (which would break a two minute throttle) and the free plan only allows 1,000 KV writes a day. The overnight save to R2 workflow is unchanged - it runs a handful of times a day. Note that the three workflows remain registered in the Cloudflare account until deleted there. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01S5mSoK49qAEDmWw1xdRJyt --- .../app/routes/uploadFromWeatherStation.ts | 19 +- .../migrations/0004_wealthy_wildside.sql | 6 + .../migrations/meta/0004_snapshot.json | 179 +++++++++++++++ .../database/migrations/meta/_journal.json | 7 + website/database/schema.d.ts | 1 + website/database/schema/UploadState.ts | 22 ++ website/worker-configuration.d.ts | 179 +++++++++++---- website/workers/app.ts | 214 +----------------- website/workers/uploads/index.ts | 130 +++++++++++ website/workers/uploads/metoffice.ts | 48 ++++ website/workers/uploads/retry.ts | 58 +++++ website/workers/uploads/windguru.ts | 43 ++++ website/workers/uploads/windy.ts | 42 ++++ website/wrangler.jsonc | 15 -- 14 files changed, 682 insertions(+), 281 deletions(-) create mode 100644 website/database/migrations/0004_wealthy_wildside.sql create mode 100644 website/database/migrations/meta/0004_snapshot.json create mode 100644 website/database/schema/UploadState.ts create mode 100644 website/workers/uploads/index.ts create mode 100644 website/workers/uploads/metoffice.ts create mode 100644 website/workers/uploads/retry.ts create mode 100644 website/workers/uploads/windguru.ts create mode 100644 website/workers/uploads/windy.ts diff --git a/website/app/routes/uploadFromWeatherStation.ts b/website/app/routes/uploadFromWeatherStation.ts index e4a21c0..a0fd489 100644 --- a/website/app/routes/uploadFromWeatherStation.ts +++ b/website/app/routes/uploadFromWeatherStation.ts @@ -4,6 +4,7 @@ import { observationFromWeatherStation, observationInsertSchema, } from "../../database/schema.d"; +import { dispatchObservation } from "../../workers/uploads"; import type { Route } from "./+types/uploadFromWeatherStation"; export async function action({ request, context, params }: Route.ActionArgs) { @@ -86,19 +87,11 @@ export async function action({ request, context, params }: Route.ActionArgs) { context.db.insert(schema.Observations).values(fullValidation.data) ); context.cloudflare.ctx.waitUntil( - context.cloudflare.env.WORKFLOW_UPLOAD_TO_METOFFICE.create({ - params: fullValidation.data, - }) - ); - context.cloudflare.ctx.waitUntil( - context.cloudflare.env.WORKFLOW_UPLOAD_TO_WINDGURU.create({ - params: fullValidation.data, - }) - ); - context.cloudflare.ctx.waitUntil( - context.cloudflare.env.WORKFLOW_UPLOAD_TO_WINDY.create({ - params: fullValidation.data, - }) + dispatchObservation( + context.cloudflare.env, + context.db, + fullValidation.data + ) ); return data({ message: "Observation received", diff --git a/website/database/migrations/0004_wealthy_wildside.sql b/website/database/migrations/0004_wealthy_wildside.sql new file mode 100644 index 0000000..91989bf --- /dev/null +++ b/website/database/migrations/0004_wealthy_wildside.sql @@ -0,0 +1,6 @@ +CREATE TABLE `upload_state` ( + `target` text PRIMARY KEY NOT NULL, + `last_attempt_at` integer, + `last_success_at` integer, + `last_error` text +); diff --git a/website/database/migrations/meta/0004_snapshot.json b/website/database/migrations/meta/0004_snapshot.json new file mode 100644 index 0000000..370be98 --- /dev/null +++ b/website/database/migrations/meta/0004_snapshot.json @@ -0,0 +1,179 @@ +{ + "version": "6", + "dialect": "sqlite", + "id": "5ec74d8e-20f9-4102-a14d-1bc93b75e84d", + "prevId": "2f57c7cd-0a6e-4818-b5e9-f7e278994bff", + "tables": { + "disregarded_observations": { + "name": "disregarded_observations", + "columns": { + "id": { + "name": "id", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": true + }, + "timestamp": { + "name": "timestamp", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "data": { + "name": "data", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "disregardReasonFriendly": { + "name": "disregardReasonFriendly", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "disregardReasonDetailed": { + "name": "disregardReasonDetailed", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "heartbeats": { + "name": "heartbeats", + "columns": { + "hour_start_timestamp": { + "name": "hour_start_timestamp", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "ping_count": { + "name": "ping_count", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "observations": { + "name": "observations", + "columns": { + "id": { + "name": "id", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": true + }, + "timestamp": { + "name": "timestamp", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "data": { + "name": "data", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "exported_to_r2": { + "name": "exported_to_r2", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "upload_state": { + "name": "upload_state", + "columns": { + "target": { + "name": "target", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "last_attempt_at": { + "name": "last_attempt_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "last_success_at": { + "name": "last_success_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "last_error": { + "name": "last_error", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + } + }, + "views": {}, + "enums": {}, + "_meta": { + "schemas": {}, + "tables": {}, + "columns": {} + }, + "internal": { + "indexes": {} + } +} \ No newline at end of file diff --git a/website/database/migrations/meta/_journal.json b/website/database/migrations/meta/_journal.json index 76855c6..5f5156e 100644 --- a/website/database/migrations/meta/_journal.json +++ b/website/database/migrations/meta/_journal.json @@ -29,6 +29,13 @@ "when": 1754915033969, "tag": "0003_powerful_dagger", "breakpoints": true + }, + { + "idx": 4, + "version": "6", + "when": 1785085307986, + "tag": "0004_wealthy_wildside", + "breakpoints": true } ] } \ No newline at end of file diff --git a/website/database/schema.d.ts b/website/database/schema.d.ts index 78f0822..4880a79 100644 --- a/website/database/schema.d.ts +++ b/website/database/schema.d.ts @@ -1,3 +1,4 @@ export * from "./schema/DisregardedObservations"; export * from "./schema/Heartbeats"; export * from "./schema/Observations"; +export * from "./schema/UploadState"; diff --git a/website/database/schema/UploadState.ts b/website/database/schema/UploadState.ts new file mode 100644 index 0000000..5169a16 --- /dev/null +++ b/website/database/schema/UploadState.ts @@ -0,0 +1,22 @@ +import { integer, sqliteTable, text } from "drizzle-orm/sqlite-core"; + +/** + * One row per external weather service we upload observations to. + * + * `lastAttemptAt` drives the per-service throttle - observations arrive from the + * weather station roughly every 30 seconds, which is far more often than the + * services will accept, so we only dispatch to a service once its minimum + * interval has elapsed. See workers/uploads/index.ts for the intervals. + * + * `lastError` exists so upload failures are still visible now that these + * uploads no longer run as Workflows (which had their own dashboard). + */ +export const UploadState = sqliteTable("upload_state", { + target: text("target").$type().primaryKey(), // The service, e.g. "windguru" + lastAttemptAt: integer("last_attempt_at", { mode: "timestamp" }), // Last time we tried, successful or not + lastSuccessAt: integer("last_success_at", { mode: "timestamp" }), // Last time an upload succeeded + lastError: text("last_error"), // Why the last attempt failed, or null if it succeeded +}); + +export const uploadTargets = ["windguru", "metoffice", "windy"] as const; +export type UploadTarget = (typeof uploadTargets)[number]; diff --git a/website/worker-configuration.d.ts b/website/worker-configuration.d.ts index 5a22737..0f3e27b 100644 --- a/website/worker-configuration.d.ts +++ b/website/worker-configuration.d.ts @@ -1,6 +1,6 @@ /* eslint-disable */ -// Generated by Wrangler by running `wrangler types` (hash: db3bd3af6c26d8b3568abd26bb3ee392) -// Runtime types generated with workerd@1.20250617.0 2025-04-05 nodejs_compat +// Generated by Wrangler by running `wrangler types` (hash: 6b253f0de7e6b493a682c1416e2aca7e) +// Runtime types generated with workerd@1.20250803.0 2025-04-05 nodejs_compat declare namespace Cloudflare { interface Env { KV: KVNamespace; @@ -8,9 +8,6 @@ declare namespace Cloudflare { DB: D1Database; CF_VERSION_METADATA: WorkerVersionMetadata; WORKFLOW_OVERNIGHT_SAVE_TO_R2: Workflow; - WORKFLOW_UPLOAD_TO_WINDGURU: Workflow; - WORKFLOW_UPLOAD_TO_METOFFICE: Workflow; - WORKFLOW_UPLOAD_TO_WINDY: Workflow; } } interface Env extends Cloudflare.Env {} @@ -353,7 +350,7 @@ interface ExecutionContext { type ExportedHandlerFetchHandler = (request: Request>, env: Env, ctx: ExecutionContext) => Response | Promise; type ExportedHandlerTailHandler = (events: TraceItem[], env: Env, ctx: ExecutionContext) => void | Promise; type ExportedHandlerTraceHandler = (traces: TraceItem[], env: Env, ctx: ExecutionContext) => void | Promise; -type ExportedHandlerTailStreamHandler = (event: TailStream.TailEvent, env: Env, ctx: ExecutionContext) => TailStream.TailEventHandlerType | Promise; +type ExportedHandlerTailStreamHandler = (event: TailStream.TailEvent, env: Env, ctx: ExecutionContext) => TailStream.TailEventHandlerType | Promise; type ExportedHandlerScheduledHandler = (controller: ScheduledController, env: Env, ctx: ExecutionContext) => void | Promise; type ExportedHandlerQueueHandler = (batch: MessageBatch, env: Env, ctx: ExecutionContext) => void | Promise; type ExportedHandlerTestHandler = (controller: TestController, env: Env, ctx: ExecutionContext) => void | Promise; @@ -1106,6 +1103,47 @@ interface ErrorEventErrorEventInit { colno?: number; error?: any; } +/** + * A message received by a target object. + * + * [MDN Reference](https://developer.mozilla.org/docs/Web/API/MessageEvent) + */ +declare class MessageEvent extends Event { + constructor(type: string, initializer: MessageEventInit); + /** + * Returns the data of the message. + * + * [MDN Reference](https://developer.mozilla.org/docs/Web/API/MessageEvent/data) + */ + readonly data: any; + /** + * Returns the origin of the message, for server-sent events and cross-document messaging. + * + * [MDN Reference](https://developer.mozilla.org/docs/Web/API/MessageEvent/origin) + */ + readonly origin: string | null; + /** + * Returns the last event ID string, for server-sent events. + * + * [MDN Reference](https://developer.mozilla.org/docs/Web/API/MessageEvent/lastEventId) + */ + readonly lastEventId: string; + /** + * Returns the WindowProxy of the source window, for cross-document messaging, and the MessagePort being attached, in the connect event fired at SharedWorkerGlobalScope objects. + * + * [MDN Reference](https://developer.mozilla.org/docs/Web/API/MessageEvent/source) + */ + readonly source: MessagePort | null; + /** + * Returns the MessagePort array sent with the message, for cross-document messaging and channel messaging. + * + * [MDN Reference](https://developer.mozilla.org/docs/Web/API/MessageEvent/ports) + */ + readonly ports: MessagePort[]; +} +interface MessageEventInit { + data: ArrayBuffer | string; +} /** * Provides a way to easily construct a set of key/value pairs representing form fields and their values, which can then be easily sent using the XMLHttpRequest.send() method. It uses the same format a form would use if the encoding type were set to "multipart/form-data". * @@ -1414,7 +1452,7 @@ interface RequestInit { signal?: (AbortSignal | null); encodeResponseBody?: "automatic" | "manual"; } -type Service = Fetcher; +type Service Rpc.WorkerEntrypointBranded) | Rpc.WorkerEntrypointBranded | ExportedHandler | undefined = undefined> = T extends new (...args: any[]) => Rpc.WorkerEntrypointBranded ? Fetcher> : T extends Rpc.WorkerEntrypointBranded ? Fetcher : T extends Exclude ? never : Fetcher; type Fetcher = (T extends Rpc.EntrypointBranded ? Rpc.Provider : unknown) & { fetch(input: RequestInfo | URL, init?: RequestInit): Promise; connect(address: SocketAddress | string, options?: SocketOptions): Socket; @@ -2287,23 +2325,6 @@ interface CloseEventInit { reason?: string; wasClean?: boolean; } -/** - * A message received by a target object. - * - * [MDN Reference](https://developer.mozilla.org/docs/Web/API/MessageEvent) - */ -declare class MessageEvent extends Event { - constructor(type: string, initializer: MessageEventInit); - /** - * Returns the data of the message. - * - * [MDN Reference](https://developer.mozilla.org/docs/Web/API/MessageEvent/data) - */ - readonly data: ArrayBuffer | string; -} -interface MessageEventInit { - data: ArrayBuffer | string; -} type WebSocketEventMap = { close: CloseEvent; message: MessageEvent; @@ -2491,6 +2512,38 @@ interface ContainerStartupOptions { enableInternet: boolean; env?: Record; } +/** + * This Channel Messaging API interface represents one of the two ports of a MessageChannel, allowing messages to be sent from one port and listening out for them arriving at the other. + * + * [MDN Reference](https://developer.mozilla.org/docs/Web/API/MessagePort) + */ +interface MessagePort extends EventTarget { + /** + * Posts a message through the channel. Objects listed in transfer are transferred, not just cloned, meaning that they are no longer usable on the sending side. + * + * Throws a "DataCloneError" DOMException if transfer contains duplicate objects or port, or if message could not be cloned. + * + * [MDN Reference](https://developer.mozilla.org/docs/Web/API/MessagePort/postMessage) + */ + postMessage(data?: any, options?: (any[] | MessagePortPostMessageOptions)): void; + /** + * Disconnects the port, so that it is no longer active. + * + * [MDN Reference](https://developer.mozilla.org/docs/Web/API/MessagePort/close) + */ + close(): void; + /** + * Begins dispatching messages received on the port. + * + * [MDN Reference](https://developer.mozilla.org/docs/Web/API/MessagePort/start) + */ + start(): void; + get onmessage(): any | null; + set onmessage(value: any | null); +} +interface MessagePortPostMessageOptions { + transfer?: any[]; +} type AiImageClassificationInput = { image: number[]; }; @@ -5266,7 +5319,7 @@ type AiModelListType = Record; declare abstract class Ai { aiGatewayLogId: string | null; gateway(gatewayId: string): AiGateway; - autorag(autoragId: string): AutoRAG; + autorag(autoragId?: string): AutoRAG; run(model: Name, inputs: InputOptions, options?: Options): Promise { * * @example 395747 */ - asn: number; + asn?: number; /** * The organization which owns the ASN of the incoming request. * * @example "Google Cloud" */ - asOrganization: string; + asOrganization?: string; /** * The original value of the `Accept-Encoding` header if Cloudflare modified it. * @@ -5910,7 +5963,7 @@ interface IncomingRequestCfPropertiesCloudflareForSaaSEnterprise { * This field is only present if you have Cloudflare for SaaS enabled on your account * and you have followed the [required steps to enable it]((https://developers.cloudflare.com/cloudflare-for-platforms/cloudflare-for-saas/domain-support/custom-metadata/)). */ - hostMetadata: HostMetadata; + hostMetadata?: HostMetadata; } interface IncomingRequestCfPropertiesCloudflareAccessOrApiShield { /** @@ -6336,6 +6389,22 @@ declare module "cloudflare:email" { }; export { _EmailMessage as EmailMessage }; } +/** + * Hello World binding to serve as an explanatory example. DO NOT USE + */ +interface HelloWorldBinding { + /** + * Retrieve the current stored value + */ + get(): Promise<{ + value: string; + ms?: number; + }>; + /** + * Set a new stored value + */ + set(value: string): Promise; +} interface Hyperdrive { /** * Connect directly to Hyperdrive as if it's your database, returning a TCP socket. @@ -6421,7 +6490,7 @@ type ImageTransform = { rotate?: 0 | 90 | 180 | 270; saturation?: number; sharpen?: number; - trim?: "border" | { + trim?: 'border' | { top?: number; bottom?: number; left?: number; @@ -6443,6 +6512,9 @@ type ImageDrawOptions = { bottom?: number; right?: number; }; +type ImageInputOptions = { + encoding?: 'base64'; +}; type ImageOutputOptions = { format: 'image/jpeg' | 'image/png' | 'image/gif' | 'image/webp' | 'image/avif' | 'rgb' | 'rgba'; quality?: number; @@ -6454,13 +6526,13 @@ interface ImagesBinding { * @throws {@link ImagesError} with code 9412 if input is not an image * @param stream The image bytes */ - info(stream: ReadableStream): Promise; + info(stream: ReadableStream, options?: ImageInputOptions): Promise; /** * Begin applying a series of transformations to an image * @param stream The image bytes * @returns A transform handle */ - input(stream: ReadableStream): ImageTransformer; + input(stream: ReadableStream, options?: ImageInputOptions): ImageTransformer; } interface ImageTransformer { /** @@ -6483,6 +6555,9 @@ interface ImageTransformer { */ output(options: ImageOutputOptions): Promise; } +type ImageTransformationOutputOptions = { + encoding?: 'base64'; +}; interface ImageTransformationResult { /** * The image as a response, ready to store in cache or return to users @@ -6495,7 +6570,7 @@ interface ImageTransformationResult { /** * The bytes of the response */ - image(): ReadableStream; + image(options?: ImageTransformationOutputOptions): ReadableStream; } interface ImagesError extends Error { readonly code: number; @@ -6720,6 +6795,19 @@ declare namespace Cloudflare { interface Env { } } +declare module 'cloudflare:node' { + export interface DefaultHandler { + fetch?(request: Request): Response | Promise; + tail?(events: TraceItem[]): void | Promise; + trace?(traces: TraceItem[]): void | Promise; + scheduled?(controller: ScheduledController): void | Promise; + queue?(batch: MessageBatch): void | Promise; + test?(controller: TestController): void | Promise; + } + export function httpServerHandler(options: { + port: number; + }, handlers?: Omit): DefaultHandler; +} declare module 'cloudflare:workers' { export type RpcStub = Rpc.Stub; export const RpcStub: { @@ -6793,6 +6881,7 @@ declare module 'cloudflare:workers' { constructor(ctx: ExecutionContext, env: Env); run(event: Readonly>, step: WorkflowStep): Promise; } + export function waitUntil(promise: Promise): void; export const env: Cloudflare.Env; } interface SecretsStoreSecret { @@ -6815,7 +6904,7 @@ declare namespace TailStream { readonly type: "fetch"; readonly method: string; readonly url: string; - readonly cfJson: string; + readonly cfJson?: object; readonly headers: Header[]; } interface JsRpcEventInfo { @@ -6926,7 +7015,7 @@ declare namespace TailStream { interface Log { readonly type: "log"; readonly level: "debug" | "error" | "info" | "log" | "warn"; - readonly message: string; + readonly message: object; } interface Return { readonly type: "return"; @@ -6947,17 +7036,27 @@ declare namespace TailStream { readonly type: "attributes"; readonly info: Attribute[]; } - interface TailEvent { - readonly traceId: string; + type EventType = Onset | Outcome | Hibernate | SpanOpen | SpanClose | DiagnosticChannelEvent | Exception | Log | Return | Link | Attributes; + interface TailEvent { readonly invocationId: string; readonly spanId: string; readonly timestamp: Date; readonly sequence: number; - readonly event: Onset | Outcome | Hibernate | SpanOpen | SpanClose | DiagnosticChannelEvent | Exception | Log | Return | Link | Attributes; + readonly event: Event; } - type TailEventHandler = (event: TailEvent) => void | Promise; - type TailEventHandlerName = "outcome" | "hibernate" | "spanOpen" | "spanClose" | "diagnosticChannel" | "exception" | "log" | "return" | "link" | "attributes"; - type TailEventHandlerObject = Record; + type TailEventHandler = (event: TailEvent) => void | Promise; + type TailEventHandlerObject = { + outcome?: TailEventHandler; + hibernate?: TailEventHandler; + spanOpen?: TailEventHandler; + spanClose?: TailEventHandler; + diagnosticChannel?: TailEventHandler; + exception?: TailEventHandler; + log?: TailEventHandler; + return?: TailEventHandler; + link?: TailEventHandler; + attributes?: TailEventHandler; + }; type TailEventHandlerType = TailEventHandler | TailEventHandlerObject; } // Copyright (c) 2022-2023 Cloudflare, Inc. diff --git a/website/workers/app.ts b/website/workers/app.ts index d69ac72..69fe387 100644 --- a/website/workers/app.ts +++ b/website/workers/app.ts @@ -1,7 +1,7 @@ import { WorkflowEntrypoint, type WorkflowEvent, - WorkflowStep, + type WorkflowStep, } from "cloudflare:workers"; import { NonRetryableError } from "cloudflare:workflows"; import { and, asc, eq, gte, lt, sql } from "drizzle-orm"; @@ -99,218 +99,6 @@ export default { }, } satisfies ExportedHandler; -export class UploadToWindGuru extends WorkflowEntrypoint< - Env, - schema.ObservationInsert -> { - async run( - event: WorkflowEvent, - step: WorkflowStep, - ) { - await step.do( - "Upload to WindGuru", - { - retries: { - limit: 2, - delay: 60000, - backoff: "exponential", - }, - timeout: "5 seconds", - }, - async () => { - const WINDGURU_UID = await this.env.KV.get("WINDGURU_UID"); - const WINDGURU_PASSWORD = await this.env.KV.get("WINDGURU_PASSWORD"); - if (!WINDGURU_UID || !WINDGURU_PASSWORD) - throw new NonRetryableError("Missing Windguru credentials"); - const payloadData = await schema.observationInsertSchema.safeParseAsync( - event.payload, - ); - if (!payloadData.success) - throw new NonRetryableError("Issue with incoming data"); - const data = payloadData.data; - - const salt = Date.now(); - const hash = await crypto.subtle.digest( - { - name: "MD5", - }, - new TextEncoder().encode( - `${salt}${WINDGURU_UID}${WINDGURU_PASSWORD}`, - ), - ); - const params = new URLSearchParams({ - uid: WINDGURU_UID, - interval: "120", - wind_avg: (data.data.wind2MinAverage / 1.151).toString(), - wind_direction: data.data.windDirection.toString(), - temperature: data.data.temperatureC.toString(), - datetime: data.timestamp.toISOString(), - salt: salt.toString(), - hash: Array.from(new Uint8Array(hash)) - .map((b) => b.toString(16).padStart(2, "0")) - .join(""), - }); - const response = await fetch( - `https://www.windguru.cz/upload/api.php?${params.toString()}`, - { - method: "GET", - }, - ).then((response) => { - if (!response.ok) - throw new Error("Windguru failed, status: " + response.status); - return response.text(); - }); - if (response !== "OK") { - throw new Error(`Windguru upload failed: ${response}`); - } - }, - ); - } -} - -export class UploadToMetOffice extends WorkflowEntrypoint< - Env, - schema.ObservationInsert -> { - async run( - event: WorkflowEvent, - step: WorkflowStep, - ) { - await step.do( - "Upload to Met Office", - { - retries: { - limit: 2, - delay: 60000, - backoff: "exponential", - }, - timeout: "5 seconds", - }, - async () => { - const METOFFICE_SITE_ID = await this.env.KV.get("METOFFICE_SITE_ID"); - const METOFFICE_AUTH_KEY = await this.env.KV.get("METOFFICE_AUTH_KEY"); - if (!METOFFICE_SITE_ID || !METOFFICE_AUTH_KEY) - throw new NonRetryableError("Missing Met Office credentials"); - const payloadData = await schema.observationInsertSchema.safeParseAsync( - event.payload, - ); - if (!payloadData.success) - throw new NonRetryableError("Issue with incoming data"); - const data = payloadData.data; - - const params = new URLSearchParams({ - siteid: METOFFICE_SITE_ID, - siteAuthenticationKey: METOFFICE_AUTH_KEY, - softwaretype: "PSC Weather Station", - dateutc: data.timestamp - .toISOString() - .replace("T", "+") - .replace(/:/g, "%3A") - .split(".")[0], - dailyrainin: (data.data.lastHourRain * 0.0393701).toString(), // real number [in]; rain inches over the past hour (alternative to precip) - dewptf: data.data.dewPoint.toString(), - tempf: data.data.temperatureF.toString(), // real number [°F]; air temperature - windspeedmph: data.data.windSpeed.toString(), // real number [mph]; wind speed (alternative to wind) - winddir: data.data.windDirection.toString(), // integer number [deg]; instantaneous wind direction - //windgustmph: data.data.windGust.toString(), // real number [mph]; current wind gust (alternative to gust) - //windgustdir: data.data.windGustDirection.toString(), // integer number [deg]; instantaneous wind direction - }); - const response = await fetch( - `http://wow.metoffice.gov.uk/automaticreading?${params.toString()}`, - { - method: "GET", - }, - ).then((response) => { - if (response.status === 429) - throw new NonRetryableError( - "Met office asked for a backoff, skip this observation", - ); - if (!response.ok) - throw new Error( - "Met Office upload failed, status: " + response.status, - ); - return response.json(); - }); - if ( - !response || - typeof response !== "object" || - Object.keys(response).length !== 0 - ) - throw new Error( - `Met Office upload failed: ${JSON.stringify(response)}`, - ); - }, - ); - } -} - -export class UploadToWindy extends WorkflowEntrypoint< - Env, - schema.ObservationInsert -> { - async run( - event: WorkflowEvent, - step: WorkflowStep, - ) { - await step.do( - "Upload to Windy", - { - retries: { - limit: 2, - delay: 60000, - backoff: "exponential", - }, - timeout: "5 seconds", - }, - async () => { - const WINDY_API_KEY = await this.env.KV.get("WINDY_API_KEY"); - const WINDY_STATION_ID = await this.env.KV.get("WINDY_STATION_ID"); - if (!WINDY_API_KEY || !WINDY_STATION_ID) - throw new NonRetryableError("Missing Windy credentials"); - const payloadData = await schema.observationInsertSchema.safeParseAsync( - event.payload, - ); - if (!payloadData.success) - throw new NonRetryableError("Issue with incoming data"); - const data = payloadData.data; - - const params = new URLSearchParams({ - station: WINDY_STATION_ID, // 32 bit integer; required for multiple stations; default value 0; alternative names: si, stationId - ts: data.timestamp.getTime().toString(), - temp: data.data.temperatureC.toString(), // real number [°C]; air temperature - windspeedmph: data.data.windSpeed.toString(), // real number [mph]; wind speed (alternative to wind) - winddir: data.data.windDirection.toString(), // integer number [deg]; instantaneous wind direction - //windgustmph: data.data.windGust.toString(), // real number [mph]; current wind gust (alternative to gust) - dewpoint: data.data.dewPoint.toString(), // real number [°C]; - precip: data.data.lastHourRain.toString(), // real number [mm]; precipitation over the past hour - }); - const response = await fetch( - `https://stations.windy.com/pws/update/${WINDY_API_KEY}?${params.toString()}`, - { - method: "GET", - }, - ); - const responseText = await response.text(); - if ( - responseText.length > 0 && - responseText.includes( - "Measurement sent too soon, update interval is 5 minutes", - ) - ) { - throw new NonRetryableError( - "Windy asked for a backoff, skip this observation", - ); - } else if (!response.ok) - throw new Error("Windy upload failed, status: " + response.status); - const responseJson = JSON.parse(responseText); - if (!responseJson || typeof responseJson !== "object") - throw new Error(`Windy upload failed: ${JSON.stringify(response)}`); - return responseText; - }, - ); - } -} - interface OvernightSaveToR2Params { dayToProcess: Date; } diff --git a/website/workers/uploads/index.ts b/website/workers/uploads/index.ts new file mode 100644 index 0000000..36a9b77 --- /dev/null +++ b/website/workers/uploads/index.ts @@ -0,0 +1,130 @@ +import { eq } from "drizzle-orm"; +import type { DrizzleD1Database } from "drizzle-orm/d1"; +import * as schema from "../../database/schema.d"; +import type { ObservationInsert, UploadTarget } from "../../database/schema.d"; +import { uploadToMetOffice } from "./metoffice"; +import { withRetry } from "./retry"; +import { uploadToWindGuru } from "./windguru"; +import { uploadToWindy } from "./windy"; + +interface UploadTargetConfig { + /** Human readable name, used in logs */ + name: string; + /** + * Shortest gap we will leave between dispatches to this service. The weather + * station reports roughly every 30 seconds, which is far more often than any + * of these services want, so most observations are skipped for most targets. + */ + minimumIntervalMs: number; + upload: (env: Env, data: ObservationInsert) => Promise; +} + +const TARGETS: Record = { + windguru: { + name: "Windguru", + minimumIntervalMs: 2 * 60 * 1000, // Matches the interval=120 we send in the payload + upload: uploadToWindGuru, + }, + metoffice: { + name: "Met Office", + minimumIntervalMs: 5 * 60 * 1000, // WOW guidance for automatic readings, and stops the 429s + upload: uploadToMetOffice, + }, + windy: { + name: "Windy", + minimumIntervalMs: 5 * 60 * 1000, // Windy's hard minimum - it rejects anything sooner + upload: uploadToWindy, + }, +}; + +/** Keep `last_error` from growing unbounded */ +const MAX_ERROR_LENGTH = 500; + +const describeError = (error: unknown) => + (error instanceof Error ? error.message : String(error)).slice( + 0, + MAX_ERROR_LENGTH, + ); + +/** + * Uploads an observation to every external service that is due one. + * + * Intended to be called from `ctx.waitUntil` so the weather station gets its + * response immediately - it can spend up to ~35 seconds retrying. The + * observation is already saved to D1 by the caller before this runs, so a + * failure here only ever costs us a data point on a third party site, and the + * next observation arrives ~30 seconds later regardless. + */ +export async function dispatchObservation( + env: Env, + db: DrizzleD1Database, + observation: ObservationInsert, +) { + try { + const now = new Date(); + const state = await db.select().from(schema.UploadState); + const lastAttemptByTarget = new Map( + state.map((row) => [row.target, row.lastAttemptAt]), + ); + + const dueTargets = schema.uploadTargets.filter((target) => { + const lastAttemptAt = lastAttemptByTarget.get(target); + if (!lastAttemptAt) return true; + return ( + now.getTime() - lastAttemptAt.getTime() >= + TARGETS[target].minimumIntervalMs + ); + }); + if (dueTargets.length === 0) { + console.debug("No upload targets are due, skipping this observation"); + return; + } + + // Claim the slot before uploading, so an observation that arrives while a + // slow retry is still running doesn't dispatch to the same service again + const [firstClaim, ...remainingClaims] = dueTargets.map((target) => + db + .insert(schema.UploadState) + .values({ target, lastAttemptAt: now }) + .onConflictDoUpdate({ + target: schema.UploadState.target, + set: { lastAttemptAt: now }, + }), + ); + await db.batch([firstClaim, ...remainingClaims]); + + const results = await Promise.allSettled( + dueTargets.map((target) => + withRetry(TARGETS[target].name, () => + TARGETS[target].upload(env, observation), + ), + ), + ); + + const [firstUpdate, ...remainingUpdates] = dueTargets.map( + (target, index) => { + const result = results[index]; + if (result.status === "fulfilled") { + console.log(`Uploaded observation to ${TARGETS[target].name}`); + return db + .update(schema.UploadState) + .set({ lastSuccessAt: now, lastError: null }) + .where(eq(schema.UploadState.target, target)); + } + console.error( + `Failed to upload observation to ${TARGETS[target].name}`, + result.reason, + ); + return db + .update(schema.UploadState) + .set({ lastError: describeError(result.reason) }) + .where(eq(schema.UploadState.target, target)); + }, + ); + await db.batch([firstUpdate, ...remainingUpdates]); + } catch (error) { + // Never let this reject - it runs detached in waitUntil and the observation + // itself has already been saved + console.error("Failed to dispatch observation to upload targets", error); + } +} diff --git a/website/workers/uploads/metoffice.ts b/website/workers/uploads/metoffice.ts new file mode 100644 index 0000000..244520a --- /dev/null +++ b/website/workers/uploads/metoffice.ts @@ -0,0 +1,48 @@ +import type { ObservationInsert } from "../../database/schema.d"; +import { NonRetryableUploadError, UPLOAD_TIMEOUT_MS } from "./retry"; + +export async function uploadToMetOffice(env: Env, data: ObservationInsert) { + const METOFFICE_SITE_ID = await env.KV.get("METOFFICE_SITE_ID"); + const METOFFICE_AUTH_KEY = await env.KV.get("METOFFICE_AUTH_KEY"); + if (!METOFFICE_SITE_ID || !METOFFICE_AUTH_KEY) + throw new NonRetryableUploadError("Missing Met Office credentials"); + + const params = new URLSearchParams({ + siteid: METOFFICE_SITE_ID, + siteAuthenticationKey: METOFFICE_AUTH_KEY, + softwaretype: "PSC Weather Station", + dateutc: data.timestamp + .toISOString() + .replace("T", "+") + .replace(/:/g, "%3A") + .split(".")[0], + dailyrainin: (data.data.lastHourRain * 0.0393701).toString(), // real number [in]; rain inches over the past hour (alternative to precip) + dewptf: data.data.dewPoint.toString(), + tempf: data.data.temperatureF.toString(), // real number [°F]; air temperature + windspeedmph: data.data.windSpeed.toString(), // real number [mph]; wind speed (alternative to wind) + winddir: data.data.windDirection.toString(), // integer number [deg]; instantaneous wind direction + //windgustmph: data.data.windGust.toString(), // real number [mph]; current wind gust (alternative to gust) + //windgustdir: data.data.windGustDirection.toString(), // integer number [deg]; instantaneous wind direction + }); + const response = await fetch( + `http://wow.metoffice.gov.uk/automaticreading?${params.toString()}`, + { + method: "GET", + signal: AbortSignal.timeout(UPLOAD_TIMEOUT_MS), + }, + ).then((response) => { + if (response.status === 429) + throw new NonRetryableUploadError( + "Met office asked for a backoff, skip this observation", + ); + if (!response.ok) + throw new Error("Met Office upload failed, status: " + response.status); + return response.json(); + }); + if ( + !response || + typeof response !== "object" || + Object.keys(response).length !== 0 + ) + throw new Error(`Met Office upload failed: ${JSON.stringify(response)}`); +} diff --git a/website/workers/uploads/retry.ts b/website/workers/uploads/retry.ts new file mode 100644 index 0000000..b99e0cf --- /dev/null +++ b/website/workers/uploads/retry.ts @@ -0,0 +1,58 @@ +/** + * Retries used to come from Cloudflare Workflows, which billed per step and so + * became expensive at one observation every ~30 seconds across three services. + * These uploads now run inline under `ctx.waitUntil`, so retries have to finish + * within the invocation that received the observation. + * + * Three attempts spread over ~18 seconds (plus up to 5 seconds per attempt + * waiting on the network) keeps the worst case around 33 seconds. That fits + * inside the shortest throttle window (2 minutes, see ./index.ts), so a slow + * retry can never overlap the next dispatch to the same service. Sleeping costs + * no CPU time, which matters on the free plan's 10ms per-invocation limit. + */ +const RETRY_DELAYS_MS = [3000, 15000]; + +/** How long a single request to an external service may take. */ +export const UPLOAD_TIMEOUT_MS = 5000; + +/** + * Thrown when retrying would be pointless or unwelcome - missing credentials, + * data the service will never accept, or the service explicitly asking us to + * back off. Replaces `NonRetryableError` from `cloudflare:workflows`. + */ +export class NonRetryableUploadError extends Error { + constructor(message: string, options?: ErrorOptions) { + super(message, options); + this.name = "NonRetryableUploadError"; + } +} + +const sleep = (ms: number) => + new Promise((resolve) => setTimeout(resolve, ms)); + +/** + * Runs `upload`, retrying transient failures. Gives up immediately on a + * NonRetryableUploadError, and rethrows the final error if every attempt fails. + */ +export async function withRetry( + name: string, + upload: () => Promise, +): Promise { + let lastError: unknown; + for (let attempt = 0; attempt <= RETRY_DELAYS_MS.length; attempt++) { + if (attempt > 0) await sleep(RETRY_DELAYS_MS[attempt - 1]); + try { + return await upload(); + } catch (error) { + lastError = error; + if (error instanceof NonRetryableUploadError) throw error; + console.warn( + `${name} upload attempt ${attempt + 1} of ${ + RETRY_DELAYS_MS.length + 1 + } failed`, + error, + ); + } + } + throw lastError; +} diff --git a/website/workers/uploads/windguru.ts b/website/workers/uploads/windguru.ts new file mode 100644 index 0000000..0a9c967 --- /dev/null +++ b/website/workers/uploads/windguru.ts @@ -0,0 +1,43 @@ +import type { ObservationInsert } from "../../database/schema.d"; +import { NonRetryableUploadError, UPLOAD_TIMEOUT_MS } from "./retry"; + +export async function uploadToWindGuru(env: Env, data: ObservationInsert) { + const WINDGURU_UID = await env.KV.get("WINDGURU_UID"); + const WINDGURU_PASSWORD = await env.KV.get("WINDGURU_PASSWORD"); + if (!WINDGURU_UID || !WINDGURU_PASSWORD) + throw new NonRetryableUploadError("Missing Windguru credentials"); + + const salt = Date.now(); + const hash = await crypto.subtle.digest( + { + name: "MD5", + }, + new TextEncoder().encode(`${salt}${WINDGURU_UID}${WINDGURU_PASSWORD}`), + ); + const params = new URLSearchParams({ + uid: WINDGURU_UID, + interval: "120", + wind_avg: (data.data.wind2MinAverage / 1.151).toString(), + wind_direction: data.data.windDirection.toString(), + temperature: data.data.temperatureC.toString(), + datetime: data.timestamp.toISOString(), + salt: salt.toString(), + hash: Array.from(new Uint8Array(hash)) + .map((b) => b.toString(16).padStart(2, "0")) + .join(""), + }); + const response = await fetch( + `https://www.windguru.cz/upload/api.php?${params.toString()}`, + { + method: "GET", + signal: AbortSignal.timeout(UPLOAD_TIMEOUT_MS), + }, + ).then((response) => { + if (!response.ok) + throw new Error("Windguru failed, status: " + response.status); + return response.text(); + }); + if (response !== "OK") { + throw new Error(`Windguru upload failed: ${response}`); + } +} diff --git a/website/workers/uploads/windy.ts b/website/workers/uploads/windy.ts new file mode 100644 index 0000000..8d68e93 --- /dev/null +++ b/website/workers/uploads/windy.ts @@ -0,0 +1,42 @@ +import type { ObservationInsert } from "../../database/schema.d"; +import { NonRetryableUploadError, UPLOAD_TIMEOUT_MS } from "./retry"; + +export async function uploadToWindy(env: Env, data: ObservationInsert) { + const WINDY_API_KEY = await env.KV.get("WINDY_API_KEY"); + const WINDY_STATION_ID = await env.KV.get("WINDY_STATION_ID"); + if (!WINDY_API_KEY || !WINDY_STATION_ID) + throw new NonRetryableUploadError("Missing Windy credentials"); + + const params = new URLSearchParams({ + station: WINDY_STATION_ID, // 32 bit integer; required for multiple stations; default value 0; alternative names: si, stationId + ts: data.timestamp.getTime().toString(), + temp: data.data.temperatureC.toString(), // real number [°C]; air temperature + windspeedmph: data.data.windSpeed.toString(), // real number [mph]; wind speed (alternative to wind) + winddir: data.data.windDirection.toString(), // integer number [deg]; instantaneous wind direction + //windgustmph: data.data.windGust.toString(), // real number [mph]; current wind gust (alternative to gust) + dewpoint: data.data.dewPoint.toString(), // real number [°C]; + precip: data.data.lastHourRain.toString(), // real number [mm]; precipitation over the past hour + }); + const response = await fetch( + `https://stations.windy.com/pws/update/${WINDY_API_KEY}?${params.toString()}`, + { + method: "GET", + signal: AbortSignal.timeout(UPLOAD_TIMEOUT_MS), + }, + ); + const responseText = await response.text(); + if ( + responseText.length > 0 && + responseText.includes( + "Measurement sent too soon, update interval is 5 minutes", + ) + ) { + throw new NonRetryableUploadError( + "Windy asked for a backoff, skip this observation", + ); + } else if (!response.ok) + throw new Error("Windy upload failed, status: " + response.status); + const responseJson = JSON.parse(responseText); + if (!responseJson || typeof responseJson !== "object") + throw new Error(`Windy upload failed: ${responseText}`); +} diff --git a/website/wrangler.jsonc b/website/wrangler.jsonc index 4930bfc..27ac56d 100644 --- a/website/wrangler.jsonc +++ b/website/wrangler.jsonc @@ -22,21 +22,6 @@ "binding": "WORKFLOW_OVERNIGHT_SAVE_TO_R2", "class_name": "OvernightSaveToR2", }, - { - "name": "PSCWeather-UploadReceivedObservationToWindGuru", - "binding": "WORKFLOW_UPLOAD_TO_WINDGURU", - "class_name": "UploadToWindGuru", - }, - { - "name": "PSCWeather-UploadReceivedObservationToMetOffice", - "binding": "WORKFLOW_UPLOAD_TO_METOFFICE", - "class_name": "UploadToMetOffice", - }, - { - "name": "PSCWeather-UploadReceivedObservationToWindy", - "binding": "WORKFLOW_UPLOAD_TO_WINDY", - "class_name": "UploadToWindy", - }, ], "r2_buckets": [ { From 660d36cc3091cbd353c6b465b24e8f7622c8f22d Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 26 Jul 2026 17:17:33 +0000 Subject: [PATCH 2/2] Fix double-encoded dateutc sent to the Met Office The dateutc value was escaped by hand and then escaped again by URLSearchParams, so what WOW actually received was the literal string "2026-07-26+12%3A00%3A00" rather than a timestamp. Pass the plain "YYYY-mm-DD HH:mm:ss" value and let URLSearchParams do the escaping, which is what WOW expects. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01S5mSoK49qAEDmWw1xdRJyt --- website/workers/uploads/metoffice.ts | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/website/workers/uploads/metoffice.ts b/website/workers/uploads/metoffice.ts index 244520a..8115f8d 100644 --- a/website/workers/uploads/metoffice.ts +++ b/website/workers/uploads/metoffice.ts @@ -11,11 +11,9 @@ export async function uploadToMetOffice(env: Env, data: ObservationInsert) { siteid: METOFFICE_SITE_ID, siteAuthenticationKey: METOFFICE_AUTH_KEY, softwaretype: "PSC Weather Station", - dateutc: data.timestamp - .toISOString() - .replace("T", "+") - .replace(/:/g, "%3A") - .split(".")[0], + // WOW wants "YYYY-mm-DD HH:mm:ss" - URLSearchParams does the escaping for + // us, so this must be the plain unescaped value + dateutc: data.timestamp.toISOString().replace("T", " ").split(".")[0], dailyrainin: (data.data.lastHourRain * 0.0393701).toString(), // real number [in]; rain inches over the past hour (alternative to precip) dewptf: data.data.dewPoint.toString(), tempf: data.data.temperatureF.toString(), // real number [°F]; air temperature