diff --git a/crates/taskito-core/BINDING_CONTRACT.md b/crates/taskito-core/BINDING_CONTRACT.md index 79aad0ab..4dcce751 100644 --- a/crates/taskito-core/BINDING_CONTRACT.md +++ b/crates/taskito-core/BINDING_CONTRACT.md @@ -17,8 +17,53 @@ depending on it). Keep Python/Node/Java specifics in the shell. them. Each shell serializes args/kwargs at enqueue and deserializes them in the worker — using whatever serializer it wants. The Python shell defaults to cloudpickle (Python-only). **Cross-language constraint:** a job enqueued by one language and run -by another requires both to agree on a neutral serializer (JSON or msgpack); -cloudpickle is Python-internal only. +by another requires both sides to use the wire envelope below. + +## Wire envelope (cross-SDK payloads) +A wire payload is one tag byte followed by the codec body. The tag records which +codec produced the body, so any shell can dispatch a decoder (or reject clearly) +without out-of-band configuration: + +| Tag | Body | Cross-SDK | Notes | +|--------|-------------|-----------|-------| +| `0x00` | native | **never** — reject with a clear error | Language-native codec (e.g. pickle). Same-language producer/consumer only. | +| `0x01` | msgpack | optional | Legacy tagged format; shells MAY read it, SHOULD NOT write it cross-SDK. | +| `0x02` | CBOR (RFC 8949) | **default** | The cross-SDK wire format. | +| `0x03` | reserved | — | Tagged JSON (not yet specified). | +| `0x04+`| reserved | — | Future (protobuf, …). | + +Untagged payloads predate the envelope and are same-SDK legacy; a shell MUST NOT +assume any tag discipline unless the task is configured with a tagged serializer +on both sides. (Sniffing is unsafe: raw msgpack/CBOR bodies can begin with any +byte value.) + +**Why CBOR over JSON**: integers survive — JS `Number` is exact only to 2^53−1 +while other languages carry 64-bit/unbounded ints; CBOR bignums round-trip them +losslessly. IANA tags also round-trip datetimes (tag 0/1), decimals (tag 4), and +byte strings without a hand-rolled registry. Mature codecs exist everywhere +(`cbor2`, `cbor-x`, `jackson-dataformat-cbor`, `fxamacker/cbor`, …). + +**Call body** (`Job.payload`): a 2-element CBOR array `[args, kwargs]` — `args` +an array, `kwargs` a map (empty map when the language has no keyword arguments). +Job-scoped extras belong in the `metadata`/`notes` columns, not the payload. +Convention for cross-SDK tasks: prefer a single object argument +(`args = [ {…} ]`, `kwargs = {}`) — it maps cleanly onto every language's +handler-binding model. + +**Result body** (`Job.result`): a bare CBOR value (no array wrapper). + +**Cross-SDK rules**: +- A shell reading tag `0x00` on a payload it did not produce MUST fail with an + error naming the tag, not a generic decode error. +- Producer and consumer of a task MUST be configured with the same wire + serializer; the tag is a self-check, not a negotiation mechanism. +- Delivery-side semantics (retries, DLQ, acks) are unaffected — the envelope is + purely a payload contract. + +**Test vectors** (hex, `0x02`-tagged CBOR): +- call `f(1, "a")`, no kwargs → `02 82 82 01 61 61 a0` — `[ [1, "a"], {} ]` +- result `true` → `02 f5` +- big int `2^53` → `02 1b 00 20 00 00 00 00 00 00` ## Dispatch call sequence 1. Shell constructs `Storage` (SQLite default; `postgres`/`redis` features) — `storage/traits.rs`. diff --git a/docs/content/docs/architecture/serialization.mdx b/docs/content/docs/architecture/serialization.mdx index ad815df2..35769c15 100644 --- a/docs/content/docs/architecture/serialization.mdx +++ b/docs/content/docs/architecture/serialization.mdx @@ -62,6 +62,7 @@ Taskito queue = Taskito.builder() | `SmartSerializer` (default) | MessagePack + cloudpickle fallback | Mixed payloads; falls back to pickle for complex objects | | `CloudpickleSerializer` | Binary (pickle) | Complex Python objects, lambdas, closures | | `JsonSerializer` | JSON | Simple types, cross-language interop, debugging | +| `CborSerializer` | CBOR (wire-envelope tag `0x02`) | Cross-SDK tasks — big integers, datetimes, bytes round-trip losslessly | @@ -71,6 +72,7 @@ Taskito queue = Taskito.builder() |---|---|---| | `JsonSerializer` (default) | JSON | Simple types, cross-language interop, debugging | | `MsgpackSerializer` | Binary (msgpack) | Compact payloads, smaller storage | +| `CborSerializer` | Binary (CBOR, wire-envelope tag `0x02`) | Cross-SDK tasks — big integers, `Date`, bytes round-trip losslessly | | `SignedSerializer` | Wraps a codec + HMAC-SHA256 | Tamper detection | | `EncryptedSerializer` | Wraps a codec + AES-256-GCM | Encrypted, authenticated payloads | @@ -82,6 +84,7 @@ Taskito queue = Taskito.builder() |---|---|---| | `JsonSerializer` (default) | JSON (Jackson) | Simple types, cross-language interop, debugging | | `MsgpackSerializer` | Binary (msgpack) | Compact payloads, smaller storage | +| `CborSerializer` | Binary (CBOR, wire-envelope tag `0x02`) | Cross-SDK tasks — big integers, decimals, bytes round-trip losslessly | | `SignedSerializer` | Wraps a serializer + HMAC-SHA256 | Tamper detection | | `EncryptedSerializer` | Wraps a serializer + AES-GCM | Encrypted, authenticated payloads | diff --git a/docs/content/docs/java/api-reference/index.mdx b/docs/content/docs/java/api-reference/index.mdx index 2782a800..e16d2320 100644 --- a/docs/content/docs/java/api-reference/index.mdx +++ b/docs/content/docs/java/api-reference/index.mdx @@ -17,7 +17,7 @@ The public API lives under `org.byteveda.taskito` (Maven coordinates | [Result](/java/api-reference/result) | Awaiting jobs and reading typed results. | | [Context](/java/api-reference/context) | `Resources.use` and the middleware `TaskContext`. | | [Resources](/java/api-reference/resources) | Worker dependency injection — scopes, pools, disposal. | -| [Serializers](/java/api-reference/serializers) | `JsonSerializer`, `MsgpackSerializer`, payload codecs. | +| [Serializers](/java/api-reference/serializers) | `JsonSerializer`, `MsgpackSerializer`, `CborSerializer`, payload codecs. | | [Batching](/java/api-reference/batching) | `Batcher` — buffer payloads and flush them as one `enqueueMany` call. | | [Workflows](/java/api-reference/workflows) | The `Workflow` builder, `WorkflowRun`, gates, sagas, analysis. | | [Canvas](/java/api-reference/canvas) | `Canvas.link`, `chain`, `group`, and `chord` shortcuts. | diff --git a/docs/content/docs/java/api-reference/overview.mdx b/docs/content/docs/java/api-reference/overview.mdx index 948ca947..3ee28c29 100644 --- a/docs/content/docs/java/api-reference/overview.mdx +++ b/docs/content/docs/java/api-reference/overview.mdx @@ -61,7 +61,7 @@ Every package lives under `org.byteveda.taskito`: | `worker` | `Worker`, `Worker.Builder`, `Handler`, `HandlerRegistry` | | `workflows` | `Workflow`, `Step`, `Canvas`, `WorkflowRun`, gates, sagas, analysis | | `resources` | Worker dependency injection — scopes, pools, `ResourceContext` | -| `serialization` | `Serializer`, `JsonSerializer`, `MsgpackSerializer`, payload codecs | +| `serialization` | `Serializer`, `JsonSerializer`, `MsgpackSerializer`, `CborSerializer`, payload codecs | | `locks` | `Lock`, `LockInfo` | | `predicates` | Enqueue-time gates — `Predicate`, `EnqueueGate`, `Recipes` | | `interception` | Enqueue interceptors — `Interceptor`, `Interception` | diff --git a/docs/content/docs/java/api-reference/serializers.mdx b/docs/content/docs/java/api-reference/serializers.mdx index 62525c1f..99642fca 100644 --- a/docs/content/docs/java/api-reference/serializers.mdx +++ b/docs/content/docs/java/api-reference/serializers.mdx @@ -1,6 +1,6 @@ --- title: Serializers -description: "JsonSerializer, MsgpackSerializer, payload codecs, and the Serializer interface." +description: "JsonSerializer, MsgpackSerializer, CborSerializer, payload codecs, and the Serializer interface." --- ```java @@ -19,12 +19,22 @@ Taskito taskito = Taskito.builder() |---|---| | `JsonSerializer` | Default. JSON via Jackson; accepts a custom `ObjectMapper`. | | `MsgpackSerializer` | Compact binary (MessagePack via Jackson). Its `org.msgpack:jackson-dataformat-msgpack` dependency is `compileOnly` — add it to your build. | +| `CborSerializer` | Binary (CBOR via Jackson), tagged with the `0x02` wire-envelope byte — the default format for tasks produced or consumed by another Taskito SDK. Its `com.fasterxml.jackson.dataformat:jackson-dataformat-cbor` dependency is `compileOnly` — add it to your build. | | `SignedSerializer(delegate, key)` | Prefixes each payload with an HMAC-SHA256 tag; verifies it (constant-time) on read — authentication, not encryption. | | `EncryptedSerializer(delegate, key)` | AES-GCM with a fresh 12-byte IV per payload — confidentiality **and** integrity. Key must be 16/24/32 bytes. | Producers and workers must configure a compatible serializer — the Rust core treats payloads as opaque bytes. +```java +import org.byteveda.taskito.serialization.CborSerializer; + +Taskito taskito = Taskito.builder() + .sqlite("taskito.db") + .serializer(new CborSerializer()) + .open(); +``` + ## Payload codecs A `PayloadCodec` is a byte-to-byte transform (`encode`/`decode`) applied after diff --git a/docs/content/docs/node/api-reference/index.mdx b/docs/content/docs/node/api-reference/index.mdx index 5277b039..0258b130 100644 --- a/docs/content/docs/node/api-reference/index.mdx +++ b/docs/content/docs/node/api-reference/index.mdx @@ -13,7 +13,7 @@ The public API exported from the `taskito` barrel. Contrib helpers live under | [Worker](/node/api-reference/worker) | `runWorker` options and the worker handle. | | [Result](/node/api-reference/result) | Awaiting results and the result waiter. | | [Context](/node/api-reference/context) | `currentJob()` — signal, progress, ids. | -| [Serializers](/node/api-reference/serializers) | `JsonSerializer`, `MsgpackSerializer`, the `Serializer` interface. | +| [Serializers](/node/api-reference/serializers) | `JsonSerializer`, `MsgpackSerializer`, `CborSerializer`, the `Serializer` interface. | | [Workflows](/node/api-reference/workflows) | The workflow builder + run handle. | | [Errors](/node/api-reference/errors) | The `TaskitoError` hierarchy. | | [CLI](/node/api-reference/cli) | The `taskito` command. | diff --git a/docs/content/docs/node/api-reference/serializers.mdx b/docs/content/docs/node/api-reference/serializers.mdx index a03cf2f8..7c2aedb5 100644 --- a/docs/content/docs/node/api-reference/serializers.mdx +++ b/docs/content/docs/node/api-reference/serializers.mdx @@ -1,6 +1,6 @@ --- title: Serializers -description: "JsonSerializer, MsgpackSerializer, and the Serializer interface." +description: "JsonSerializer, MsgpackSerializer, CborSerializer, and the Serializer interface." --- ```ts @@ -14,6 +14,7 @@ import type { Serializer } from "@byteveda/taskito"; |---|---| | `JsonSerializer` | Default. Human-readable JSON bytes. | | `MsgpackSerializer` | Compact binary (msgpack). | +| `CborSerializer` | Binary (CBOR, RFC 8949), tagged with the `0x02` wire-envelope byte — the default format for tasks produced or consumed by another Taskito SDK. Integers beyond `Number.MAX_SAFE_INTEGER` round-trip as `BigInt`. | | `SignedSerializer` | Wraps another codec with an HMAC-SHA256 tag; rejects tampered payloads. | | `EncryptedSerializer` | Wraps another codec with AES-256-GCM; encrypts + authenticates. | @@ -21,6 +22,12 @@ import type { Serializer } from "@byteveda/taskito"; new Queue({ dbPath: "taskito.db", serializer: new MsgpackSerializer() }); ``` +```ts +import { Queue, CborSerializer } from "@byteveda/taskito"; + +new Queue({ dbPath: "taskito.db", serializer: new CborSerializer() }); +``` + `SignedSerializer(secret, inner?)` prefixes each payload with a 32-byte HMAC tag and verifies it (timing-safe) on read — authentication, not encryption. `EncryptedSerializer(secret, inner?)` goes further: AES-256-GCM gives diff --git a/docs/content/docs/shared/guides/extensibility/serializers.mdx b/docs/content/docs/shared/guides/extensibility/serializers.mdx index 995b669b..2333b34f 100644 --- a/docs/content/docs/shared/guides/extensibility/serializers.mdx +++ b/docs/content/docs/shared/guides/extensibility/serializers.mdx @@ -1,6 +1,6 @@ --- title: Pluggable Serializers -description: "Built-in JSON, MessagePack, signed, and encrypted serializers, plus the cross-SDK payload-codec chain for compression and encryption." +description: "Built-in JSON, MessagePack, CBOR, signed, and encrypted serializers, plus the cross-SDK payload-codec chain for compression and encryption." --- Task arguments and results are serialized with a pluggable `Serializer`. The @@ -26,15 +26,16 @@ strings, numbers, booleans, `None`, tuples), with an automatic The default is `JsonSerializer` — human-readable and easy to debug. Other -built-ins: `MsgpackSerializer`, `SignedSerializer`, `EncryptedSerializer`. +built-ins: `MsgpackSerializer`, `CborSerializer`, `SignedSerializer`, +`EncryptedSerializer`. The default is `JsonSerializer` — Jackson-backed, human-readable and easy to -debug. Other built-ins: `MsgpackSerializer`, `SignedSerializer`, -`EncryptedSerializer`. +debug. Other built-ins: `MsgpackSerializer`, `CborSerializer`, +`SignedSerializer`, `EncryptedSerializer`. @@ -176,6 +177,77 @@ you want to spell it out in your own requirements file. +### CborSerializer + +Binary serialization writing the `0x02` wire-envelope tag — the default +format for tasks produced or consumed by another Taskito SDK. Unlike JSON, +big integers, `datetime`/`Date`, `bytes`, and decimals round-trip losslessly +across languages. + + + + +```python +from taskito import CborSerializer, Queue + +queue = Queue(serializer=CborSerializer()) +``` + + + + +```ts +import { Queue, CborSerializer } from "@byteveda/taskito"; + +new Queue({ dbPath: "taskito.db", serializer: new CborSerializer() }); +``` + + + + +```java +import org.byteveda.taskito.Taskito; +import org.byteveda.taskito.serialization.CborSerializer; + +Taskito taskito = Taskito.builder() + .sqlite("taskito.db") + .serializer(new CborSerializer()) + .open(); +``` + + + + + + +`cbor2` ships as a required taskito dependency — no extra install needed. The +default `SmartSerializer` can already *read* `0x02`-tagged CBOR payloads (for +example one written by another Taskito SDK) without extra configuration; use +`CborSerializer` explicitly to *write* cross-SDK payloads. + + + + + +`cbor-x` ships as a core dependency — no extra install required. Integers +beyond `Number.MAX_SAFE_INTEGER` round-trip as `BigInt`. + + + + + +`CborSerializer`'s `com.fasterxml.jackson.dataformat:jackson-dataformat-cbor` +dependency is `compileOnly` — add it to your build explicitly, same as +`MsgpackSerializer`. + + + + + Producer and consumer of a task must configure the same wire serializer. + For cross-SDK tasks, prefer a single object/dict argument — it maps + cleanly onto every language's handler-binding model. + + ### SignedSerializer HMAC-SHA256 integrity tag. Unlike `EncryptedSerializer`, it does **not** hide @@ -411,21 +483,23 @@ override it in a generics-aware implementation, as `JsonSerializer` does. ## Choosing a serializer -| | SmartSerializer | CloudpickleSerializer | JsonSerializer | MsgPackSerializer | EncryptedSerializer | -|---|---|---|---|---|---| -| **Complex objects** | Yes (cloudpickle fallback) | Yes | No | No | Depends on inner serializer | -| **Debugging** | Binary payloads (opaque) | Binary payloads (opaque) | Human-readable JSON | Binary (opaque) | Ciphertext (opaque) | -| **Cross-language** | Python only | Python only | Any language | Any language | Python only (by default) | -| **Performance** | Best for plain data | Good | Good for simple types | Best | Adds encryption overhead | -| **Security** | None | None | None | None | AES-256-GCM | -| **Extra dependency** | No | No | No | No | `cryptography` (`taskito[encryption]`) | -| **Default** | Yes | No | No | No | No | +| | SmartSerializer | CloudpickleSerializer | JsonSerializer | MsgPackSerializer | CborSerializer | EncryptedSerializer | +|---|---|---|---|---|---|---| +| **Complex objects** | Yes (cloudpickle fallback) | Yes | No | No | No (CBOR-representable types only) | Depends on inner serializer | +| **Debugging** | Binary payloads (opaque) | Binary payloads (opaque) | Human-readable JSON | Binary (opaque) | Binary (opaque) | Ciphertext (opaque) | +| **Cross-language** | Python only | Python only | Portable format, same-SDK payloads | Portable format, same-SDK payloads | Yes (cross-SDK wire format) | Python only (by default) | +| **Performance** | Best for plain data | Good | Good for simple types | Best | Good | Adds encryption overhead | +| **Security** | None | None | None | None | None | AES-256-GCM | +| **Extra dependency** | No | No | No | No | No | `cryptography` (`taskito[encryption]`) | +| **Default** | Yes | No | No | No | No | No | **Rule of thumb**: use the default `SmartSerializer` unless you have a specific reason to switch — it gets you compact MessagePack payloads for plain data with an automatic cloudpickle fallback for anything MessagePack -can't encode. Use `EncryptedSerializer` when tasks carry sensitive data that -must not be readable in the database. See +can't encode. Use `CborSerializer` when a task is produced or consumed by +another Taskito SDK — it's the only serializer here whose wire format is +part of the cross-SDK contract. Use `EncryptedSerializer` when tasks carry +sensitive data that must not be readable in the database. See [Architecture: Serialization](/architecture/serialization) for a shorter overview and what gets serialized where. diff --git a/sdks/java/build.gradle.kts b/sdks/java/build.gradle.kts index aaf46026..eab11896 100644 --- a/sdks/java/build.gradle.kts +++ b/sdks/java/build.gradle.kts @@ -91,6 +91,11 @@ dependencies { compileOnly("org.msgpack:jackson-dataformat-msgpack:0.9.8") testImplementation("org.msgpack:jackson-dataformat-msgpack:0.9.8") + // Optional: the CBOR wire serializer (cross-SDK payloads). Same model — + // consumers that use CborSerializer add this dependency themselves. + compileOnly("com.fasterxml.jackson.dataformat:jackson-dataformat-cbor:2.17.2") + testImplementation("com.fasterxml.jackson.dataformat:jackson-dataformat-cbor:2.17.2") + // Optional: observability contrib middleware. Consumers that use them add the // matching runtime dependency themselves. compileOnly("io.micrometer:micrometer-observation:1.13.6") diff --git a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java index 1d2dd0a8..a993ade1 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java @@ -262,7 +262,7 @@ private Optional dispatchEnqueue( // Serialize before codec-encoding so the idempotency key hashes the deterministic // pre-codec payload — a non-deterministic codec (e.g. an AES-GCM nonce) must not // change the dedup key. - byte[] payloadBytes = serializer.serialize(context.payload()); + byte[] payloadBytes = serializer.serializeCall(context.payload()); String uniqueKey = resolveUniqueKey(taskName, payloadBytes, finalOptions, taskIdempotentDefault); if (uniqueKey != null && !uniqueKey.equals(finalOptions.uniqueKey())) { finalOptions = finalOptions.toBuilder().uniqueKey(uniqueKey).build(); @@ -359,7 +359,7 @@ public List enqueueMany(Task task, List payloads, EnqueueOptio // auto-derive, mirroring the single-enqueue path (explicit // uniqueKey/idempotencyKey precedence preserved). Key hashes the // deterministic pre-codec bytes. - byte[] payloadBytes = serializer.serialize(payload); + byte[] payloadBytes = serializer.serializeCall(payload); EnqueueOptions jobOptions = options; String uniqueKey = resolveUniqueKey(task.name(), payloadBytes, jobOptions, task.idempotent()); if (uniqueKey != null && !uniqueKey.equals(jobOptions.uniqueKey())) { @@ -634,7 +634,7 @@ public Optional getLockInfo(String name) { @Override public long registerPeriodic(PeriodicTask task) { - byte[] payload = task.payload == null ? null : serializer.serialize(task.payload); + byte[] payload = task.payload == null ? null : serializer.serializeCall(task.payload); return backend.registerPeriodic( task.name, task.taskName, task.cron, payload, task.queue, task.timezone, task.enabled); } @@ -688,7 +688,7 @@ public WorkflowRun submitWorkflow(Workflow workflow, Map supplie + "' has no payload; supply one via submitWorkflow(wf, payloads)"); } payloadNames.add(step.name); - payloads.add(serializer.serialize(payload)); + payloads.add(serializer.serializeCall(payload)); } } String runId = backend.submitWorkflow( @@ -869,7 +869,7 @@ private String encodeChild(Workflow child) { } specs.add(stepSpec(step)); if (step.payload != null) { - payloads.put(step.name, Base64.getEncoder().encodeToString(serializer.serialize(step.payload))); + payloads.put(step.name, Base64.getEncoder().encodeToString(serializer.serializeCall(step.payload))); } } Map blob = new LinkedHashMap<>(); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/serialization/CborSerializer.java b/sdks/java/src/main/java/org/byteveda/taskito/serialization/CborSerializer.java new file mode 100644 index 00000000..3c303d0c --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/serialization/CborSerializer.java @@ -0,0 +1,116 @@ +package org.byteveda.taskito.serialization; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.dataformat.cbor.databind.CBORMapper; +import java.lang.reflect.Type; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import org.byteveda.taskito.errors.SerializationException; + +/** + * CBOR serializer for cross-SDK payloads (RFC 8949), writing the {@code 0x02} + * wire-envelope tag from the binding contract. Use for tasks produced or + * consumed by another Taskito SDK: unlike JSON, CBOR round-trips 64-bit and + * larger integers, {@code byte[]}, and decimals losslessly across languages. + * + *

Call payloads use the cross-SDK call body {@code [args, kwargs]} with the + * task payload as the single positional argument. Requires the optional + * {@code com.fasterxml.jackson.dataformat:jackson-dataformat-cbor} dependency. + */ +public final class CborSerializer implements Serializer { + private static final byte TAG_NATIVE = 0x00; + private static final byte TAG_CBOR = 0x02; + + private final ObjectMapper mapper; + + public CborSerializer() { + this(CBORMapper.builder().build()); + } + + public CborSerializer(ObjectMapper mapper) { + this.mapper = mapper; + } + + @Override + public byte[] serialize(Object value) { + try { + return tagged(mapper.writeValueAsBytes(value)); + } catch (Exception e) { + throw new SerializationException("failed to serialize payload", e); + } + } + + @Override + public T deserialize(byte[] bytes, Class type) { + try { + return mapper.readValue(body(bytes), type); + } catch (SerializationException e) { + throw e; + } catch (Exception e) { + throw new SerializationException("failed to deserialize payload", e); + } + } + + @Override + public Object deserialize(byte[] bytes, Type type) { + try { + return mapper.readValue(body(bytes), mapper.getTypeFactory().constructType(type)); + } catch (SerializationException e) { + throw e; + } catch (Exception e) { + throw new SerializationException("failed to deserialize payload", e); + } + } + + @Override + public byte[] serializeCall(Object payload) { + List call = Arrays.asList(Arrays.asList(payload), Collections.emptyMap()); + try { + return tagged(mapper.writeValueAsBytes(call)); + } catch (Exception e) { + throw new SerializationException("failed to serialize call payload", e); + } + } + + @Override + public Object deserializeCall(byte[] bytes, Type payloadType) { + try { + JsonNode call = mapper.readTree(body(bytes)); + if (!call.isArray() || call.size() != 2 || !call.get(0).isArray()) { + throw new SerializationException("CBOR call payload is not the [args, kwargs] wire shape"); + } + JsonNode args = call.get(0); + JsonNode first = + args.size() > 0 ? args.get(0) : mapper.getNodeFactory().nullNode(); + return mapper.convertValue(first, mapper.getTypeFactory().constructType(payloadType)); + } catch (SerializationException e) { + throw e; + } catch (Exception e) { + throw new SerializationException("failed to deserialize call payload", e); + } + } + + private static byte[] tagged(byte[] cborBody) { + byte[] out = new byte[cborBody.length + 1]; + out[0] = TAG_CBOR; + System.arraycopy(cborBody, 0, out, 1, cborBody.length); + return out; + } + + private static byte[] body(byte[] bytes) { + if (bytes == null || bytes.length == 0) { + throw new SerializationException("cannot deserialize an empty payload"); + } + if (bytes[0] == TAG_CBOR) { + return Arrays.copyOfRange(bytes, 1, bytes.length); + } + if (bytes[0] == TAG_NATIVE) { + throw new SerializationException("payload is native-tagged (0x00): produced by a" + + " same-language-only serializer, not readable as CBOR wire format"); + } + throw new SerializationException( + String.format("payload is not CBOR wire format (tag 0x%02x, expected 0x02)", bytes[0])); + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/serialization/Serializer.java b/sdks/java/src/main/java/org/byteveda/taskito/serialization/Serializer.java index f3125088..c065dfd2 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/serialization/Serializer.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/serialization/Serializer.java @@ -22,4 +22,19 @@ default Object deserialize(byte[] bytes, Type type) { + " does not support the generic payload type " + type + "; use a Jackson-based serializer or a non-generic Task payload type"); } + + /** + * Call-shaped encoding for task payloads. Wire serializers (e.g. + * {@link CborSerializer}) override this to write the cross-SDK call body + * {@code [args, kwargs]} from the binding contract; the default keeps the + * bare-value body. Results always use {@link #serialize}/{@link #deserialize}. + */ + default byte[] serializeCall(Object payload) { + return serialize(payload); + } + + /** Inverse of {@link #serializeCall}: decode a call payload to the handler argument. */ + default Object deserializeCall(byte[] bytes, Type payloadType) { + return deserialize(bytes, payloadType); + } } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java b/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java index 69175a35..5bc6dc35 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java @@ -93,7 +93,7 @@ private void runJob(long token, String jobId, String taskName, byte[] payload) { for (Middleware m : middleware) { m.before(context); } - Object argument = serializer.deserialize(decodePayload(payload, task.codecs), task.payloadType); + Object argument = serializer.deserializeCall(decodePayload(payload, task.codecs), task.payloadType); Object result = task.handler.apply(argument); for (Middleware m : middleware) { m.after(context, result); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java index a1aa6836..f20d8d5e 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java @@ -93,7 +93,7 @@ public void register(Workflow workflow) { Map conditions = new HashMap<>(); for (Step step : workflow.steps()) { if (step.payload != null) { - byNode.put(step.name, serializer.serialize(step.payload)); + byNode.put(step.name, serializer.serializeCall(step.payload)); } if (step.callableCondition != null) { conditions.put(step.name, step.callableCondition); @@ -196,7 +196,7 @@ private void expandFanOut(String runId, PlanNode fanOut, List items, RunPlan byte[][] payloads = new byte[items.size()][]; for (int i = 0; i < items.size(); i++) { names[i] = fanOut.name + "[" + i + "]"; - payloads[i] = serializer.serialize(items.get(i)); + payloads[i] = serializer.serializeCall(items.get(i)); } backend.expandFanOut( runId, @@ -253,7 +253,7 @@ private void collectFanIns(String runId, String parent, List results, Ru finalizeIfTerminal(runId); return; } - byte[] payload = serializer.serialize(results); + byte[] payload = serializer.serializeCall(results); for (PlanNode fanIn : fanIns) { backend.createDeferredJob( runId, diff --git a/sdks/java/src/test/java/org/byteveda/taskito/serialization/CborSerializerTest.java b/sdks/java/src/test/java/org/byteveda/taskito/serialization/CborSerializerTest.java new file mode 100644 index 00000000..2aeb5087 --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/serialization/CborSerializerTest.java @@ -0,0 +1,115 @@ +package org.byteveda.taskito.serialization; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.math.BigInteger; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import org.byteveda.taskito.errors.SerializationException; +import org.junit.jupiter.api.Test; + +class CborSerializerTest { + + private final CborSerializer cbor = new CborSerializer(); + + @Test + void roundTripsScalar() { + byte[] bytes = cbor.serialize(42); + assertEquals(42, cbor.deserialize(bytes, Integer.class)); + } + + @Test + @SuppressWarnings("unchecked") + void roundTripsMap() { + Map value = new LinkedHashMap<>(); + value.put("count", 3); + value.put("name", "taskito"); + + Map back = cbor.deserialize(cbor.serialize(value), Map.class); + + assertEquals(3, ((Number) back.get("count")).intValue()); + assertEquals("taskito", back.get("name")); + } + + @Test + void writesWireTag() { + assertEquals(0x02, cbor.serialize(List.of(1, 2))[0]); + assertEquals(0x02, cbor.serializeCall("payload")[0]); + } + + @Test + void roundTripsLongAndBigInteger() { + long big = (1L << 53) + 1; + assertEquals(big, cbor.deserialize(cbor.serialize(big), Long.class)); + BigInteger huge = BigInteger.TWO.pow(80); + assertEquals(huge, cbor.deserialize(cbor.serialize(huge), BigInteger.class)); + } + + @Test + void encodesCallsAsArgsKwargsWireShape() { + byte[] bytes = cbor.serializeCall("value"); + List call = cbor.deserialize(bytes, List.class); + assertEquals(2, call.size()); + assertEquals(List.of("value"), call.get(0)); + assertEquals(Map.of(), call.get(1)); + } + + @Test + void deserializeCallReturnsSinglePayloadArgument() { + byte[] bytes = cbor.serializeCall("hello"); + assertEquals("hello", cbor.deserializeCall(bytes, String.class)); + } + + @Test + void deserializeCallHandlesNullPayload() { + byte[] bytes = cbor.serializeCall(null); + assertEquals(null, cbor.deserializeCall(bytes, String.class)); + } + + @Test + void rejectsNativeTaggedPayload() { + SerializationException error = assertThrows( + SerializationException.class, () -> cbor.deserialize(new byte[] {0x00, (byte) 0x80}, Object.class)); + assertTrue(error.getMessage().contains("native-tagged")); + } + + @Test + void rejectsUntaggedPayload() { + SerializationException error = assertThrows( + SerializationException.class, () -> cbor.deserialize("{\"json\":true}".getBytes(), Object.class)); + assertTrue(error.getMessage().contains("not CBOR wire format")); + } + + @Test + void rejectsEmptyPayload() { + assertThrows(SerializationException.class, () -> cbor.deserialize(new byte[0], Object.class)); + } + + @Test + void rejectsNonCallShapeInDeserializeCall() { + byte[] notACall = cbor.serialize(Map.of("not", "a call")); + SerializationException error = + assertThrows(SerializationException.class, () -> cbor.deserializeCall(notACall, String.class)); + assertTrue(error.getMessage().contains("wire shape")); + } + + @Test + void decodesBindingContractVector() { + // BINDING_CONTRACT.md test vector for call f(1, "a"): [[1, "a"], {}] + byte[] vector = {0x02, (byte) 0x82, (byte) 0x82, 0x01, 0x61, 0x61, (byte) 0xa0}; + List call = cbor.deserialize(vector, List.class); + assertEquals(List.of(1, "a"), call.get(0)); + assertEquals(Map.of(), call.get(1)); + } + + @Test + void defaultSerializerCallShapeIsUnchanged() { + Serializer json = new JsonSerializer(); + byte[] bytes = json.serializeCall("plain"); + assertEquals("plain", json.deserializeCall(bytes, String.class)); + assertEquals("plain", json.deserialize(bytes, String.class)); + } +} diff --git a/sdks/node/package.json b/sdks/node/package.json index 9ea169ec..71b8dc00 100644 --- a/sdks/node/package.json +++ b/sdks/node/package.json @@ -120,6 +120,7 @@ }, "dependencies": { "@msgpack/msgpack": "^3.1.3", + "cbor-x": "^1.6.4", "commander": "^15.0.0" }, "peerDependencies": { diff --git a/sdks/node/pnpm-lock.yaml b/sdks/node/pnpm-lock.yaml index 372b15b6..872b9f77 100644 --- a/sdks/node/pnpm-lock.yaml +++ b/sdks/node/pnpm-lock.yaml @@ -11,6 +11,9 @@ importers: '@msgpack/msgpack': specifier: ^3.1.3 version: 3.1.3 + cbor-x: + specifier: ^1.6.4 + version: 1.6.4 commander: specifier: ^15.0.0 version: 15.0.0 @@ -129,6 +132,36 @@ packages: '@borewit/text-codec@0.2.2': resolution: {integrity: sha512-DDaRehssg1aNrH4+2hnj1B7vnUGEjU6OIlyRdkMd0aUdIUvKXrJfXsy8LVtXAy7DRvYVluWbMspsRhz2lcW0mQ==} + '@cbor-extract/cbor-extract-darwin-arm64@2.2.2': + resolution: {integrity: sha512-ZKZ/F8US7JR92J4DMct6cLW/Y66o2K576+zjlEN/MevH70bFIsB10wkZEQPLzl2oNh2SMGy55xpJ9JoBRl5DOA==} + cpu: [arm64] + os: [darwin] + + '@cbor-extract/cbor-extract-darwin-x64@2.2.2': + resolution: {integrity: sha512-32b1mgc+P61Js+KW9VZv/c+xRw5EfmOcPx990JbCBSkYJFY0l25VinvyyWfl+3KjibQmAcYwmyzKF9J4DyKP/Q==} + cpu: [x64] + os: [darwin] + + '@cbor-extract/cbor-extract-linux-arm64@2.2.2': + resolution: {integrity: sha512-wfqgzqCAy/Vn8i6WVIh7qZd0DdBFaWBjPdB6ma+Wihcjv0gHqD/mw3ouVv7kbbUNrab6dKEx/w3xQZEdeXIlzg==} + cpu: [arm64] + os: [linux] + + '@cbor-extract/cbor-extract-linux-arm@2.2.2': + resolution: {integrity: sha512-tNg0za41TpQfkhWjptD+0gSD2fggMiDCSacuIeELyb2xZhr7PrhPe5h66Jc67B/5dmpIhI2QOUtv4SBsricyYQ==} + cpu: [arm] + os: [linux] + + '@cbor-extract/cbor-extract-linux-x64@2.2.2': + resolution: {integrity: sha512-rpiLnVEsqtPJ+mXTdx1rfz4RtUGYIUg2rUAZgd1KjiC1SehYUSkJN7Yh+aVfSjvCGtVP0/bfkQkXpPXKbmSUaA==} + cpu: [x64] + os: [linux] + + '@cbor-extract/cbor-extract-win32-x64@2.2.2': + resolution: {integrity: sha512-dI+9P7cfWxkTQ+oE+7Aa6onEn92PHgfWXZivjNheCRmTBDBf2fx6RyTi0cmgpYLnD1KLZK9ZYrMxaPZ4oiXhGA==} + cpu: [x64] + os: [win32] + '@emnapi/core@1.10.0': resolution: {integrity: sha512-yq6OkJ4p82CAfPl0u9mQebQHKPJkY7WrIuk205cTYnYe+k2Z8YBh11FrbRG/H6ihirqcacOgl2BIO8oyMQLeXw==} @@ -863,6 +896,13 @@ packages: resolution: {integrity: sha512-+ys997U96po4Kx/ABpBCqhA9EuxJaQWDQg7295H4hBphv3IZg0boBKuwYpt4YXp6MZ5AmZQnU/tyMTlRpaSejg==} engines: {node: '>= 0.4'} + cbor-extract@2.2.2: + resolution: {integrity: sha512-hlSxxI9XO2yQfe9g6msd3g4xCfDqK5T5P0fRMLuaLHhxn4ViPrm+a+MUfhrvH2W962RGxcBwEGzLQyjbDG1gng==} + hasBin: true + + cbor-x@1.6.4: + resolution: {integrity: sha512-UGKHjp6RHC6QuZ2yy5LCKm7MojM4716DwoSaqwQpaH4DvZvbBTGcoDNTiG9Y2lByXZYFEs9WRkS5tLl96IrF1Q==} + chai@6.2.2: resolution: {integrity: sha512-NUPRluOfOiTKBKvWPtSD4PhFvWCqOi0BGStNWs57X9js7XGTprSmFoz5F0tWhR4WPjNeR9jXqdC7/UpSJTnlRg==} engines: {node: '>=18'} @@ -1249,6 +1289,10 @@ packages: resolution: {integrity: sha512-8Ofs/AUQh8MaEcrlq5xOX0CQ9ypTF5dl78mjlMNfOK08fzpgTHQRQPBxcPlEtIw0yRpws+Zo/3r+5WRby7u3Gg==} engines: {node: '>= 0.6'} + node-gyp-build-optional-packages@5.1.1: + resolution: {integrity: sha512-+P72GAjVAbTxjjwUmwjVrqrdZROD4nf8KgpBoDxqXXTiYZZt/ud60dE5yvCSr9lRO8e8yv6kgJIC0K0PfZFVQw==} + hasBin: true + object-assign@4.1.1: resolution: {integrity: sha512-rJgTQnkUnH1sFw8yT6VSU3zD3sWmu6sZhIseY8VX+GRu3P6F7Fu+JNDoXfklElbLJSnc3FUQHVe4cU5hj+BcUg==} engines: {node: '>=0.10.0'} @@ -1732,6 +1776,24 @@ snapshots: '@borewit/text-codec@0.2.2': {} + '@cbor-extract/cbor-extract-darwin-arm64@2.2.2': + optional: true + + '@cbor-extract/cbor-extract-darwin-x64@2.2.2': + optional: true + + '@cbor-extract/cbor-extract-linux-arm64@2.2.2': + optional: true + + '@cbor-extract/cbor-extract-linux-arm@2.2.2': + optional: true + + '@cbor-extract/cbor-extract-linux-x64@2.2.2': + optional: true + + '@cbor-extract/cbor-extract-win32-x64@2.2.2': + optional: true + '@emnapi/core@1.10.0': dependencies: '@emnapi/wasi-threads': 1.2.1 @@ -2289,6 +2351,22 @@ snapshots: call-bind-apply-helpers: 1.0.2 get-intrinsic: 1.3.0 + cbor-extract@2.2.2: + dependencies: + node-gyp-build-optional-packages: 5.1.1 + optionalDependencies: + '@cbor-extract/cbor-extract-darwin-arm64': 2.2.2 + '@cbor-extract/cbor-extract-darwin-x64': 2.2.2 + '@cbor-extract/cbor-extract-linux-arm': 2.2.2 + '@cbor-extract/cbor-extract-linux-arm64': 2.2.2 + '@cbor-extract/cbor-extract-linux-x64': 2.2.2 + '@cbor-extract/cbor-extract-win32-x64': 2.2.2 + optional: true + + cbor-x@1.6.4: + optionalDependencies: + cbor-extract: 2.2.2 + chai@6.2.2: {} chokidar@4.0.3: @@ -2674,6 +2752,11 @@ snapshots: negotiator@1.0.0: {} + node-gyp-build-optional-packages@5.1.1: + dependencies: + detect-libc: 2.1.2 + optional: true + object-assign@4.1.1: {} object-inspect@1.13.4: {} diff --git a/sdks/node/src/index.ts b/sdks/node/src/index.ts index cf2f89e2..f50987f2 100644 --- a/sdks/node/src/index.ts +++ b/sdks/node/src/index.ts @@ -64,6 +64,7 @@ export { export { type ScalerOptions, serveScaler } from "./scaler"; export { AesGcmCodec, + CborSerializer, CodecSerializer, EncryptedSerializer, GzipCodec, diff --git a/sdks/node/src/queue.ts b/sdks/node/src/queue.ts index e1ea9e65..599c75b3 100644 --- a/sdks/node/src/queue.ts +++ b/sdks/node/src/queue.ts @@ -40,7 +40,13 @@ import { ResourceRuntime, type ResourceScope, } from "./resources"; -import { CodecSerializer, JsonSerializer, type PayloadCodec, type Serializer } from "./serializers"; +import { + CodecSerializer, + JsonSerializer, + type PayloadCodec, + type Serializer, + serializeCall, +} from "./serializers"; import type { AnyHandler, CircuitBreaker, @@ -159,7 +165,9 @@ export class Queue { if (typeof this.native.markWorkflowNodeResult !== "function") { return undefined; } - this.workflowTracker ??= new WorkflowTracker(this.native, this.serializer); + this.workflowTracker ??= new WorkflowTracker(this.native, this.serializer, (taskName, args) => + this.encodeTaskPayload(taskName, args), + ); return this.workflowTracker; } @@ -494,11 +502,12 @@ export class Queue { } /** - * Serialize a task payload and apply the task's named codecs in order. - * Payload only — results stay on the queue serializer. + * Serialize task args (call-shaped, honoring wire serializers) and apply + * the task's named codecs in order. Payload only — results stay on the + * queue serializer's plain `serialize`. */ - private encodeTaskPayload(taskName: string, value: unknown): Buffer { - let data = this.serializer.serialize(value); + private encodeTaskPayload(taskName: string, args: unknown): Buffer { + let data = serializeCall(this.serializer, Array.isArray(args) ? args : [args]); for (const name of this.tasks.get(taskName)?.options?.codecs ?? []) { const codec = this.codecs.get(name); if (!codec) { diff --git a/sdks/node/src/serializers/cbor.ts b/sdks/node/src/serializers/cbor.ts new file mode 100644 index 00000000..4bc49abc --- /dev/null +++ b/sdks/node/src/serializers/cbor.ts @@ -0,0 +1,82 @@ +import { Decoder, Encoder } from "cbor-x"; +import { SerializationError } from "../errors"; +import type { Serializer } from "./serializer"; + +/** Wire-envelope tags (see taskito-core BINDING_CONTRACT.md "Wire envelope"). */ +const TAG_NATIVE = 0x00; +const TAG_CBOR = 0x02; + +function tagged(body: Uint8Array): Uint8Array { + const out = new Uint8Array(body.length + 1); + out[0] = TAG_CBOR; + out.set(body, 1); + return out; +} + +/** + * CBOR serializer for cross-SDK payloads (RFC 8949), writing the `0x02` + * wire-envelope tag. Use for tasks produced or consumed by another Taskito + * SDK: unlike JSON, CBOR round-trips integers beyond `Number.MAX_SAFE_INTEGER` + * (as `BigInt`), `Date` (tag 1), and binary data losslessly across languages. + * + * Call payloads use the cross-SDK call body `[args, kwargs]`; a producer with + * keyword arguments surfaces them here as a trailing options object. + */ +export class CborSerializer implements Serializer { + // Plain RFC 8949 structures only: cbor-x "records" and tag-259 maps are + // extensions other SDKs cannot decode (tag 259 defaults ON once records are + // off). `variableMapSize` emits minimal (canonical) map headers so output + // matches other SDKs' encoders. + // `useTag259ForMaps` is honored at runtime but missing from cbor-x's typings. + private readonly encoder = new Encoder({ + useRecords: false, + useTag259ForMaps: false, + variableMapSize: true, + } as ConstructorParameters[0]); + private readonly decoder = new Decoder({ useRecords: false, mapsAsObjects: true }); + + serialize(value: unknown): Uint8Array { + return tagged(this.encoder.encode(value ?? null)); + } + + deserialize(bytes: Uint8Array): unknown { + return this.decoder.decode(this.body(bytes)); + } + + serializeCall(args: unknown[]): Uint8Array { + return tagged(this.encoder.encode([args, {}])); + } + + deserializeCall(bytes: Uint8Array): unknown[] { + const decoded = this.decoder.decode(this.body(bytes)); + if ( + !Array.isArray(decoded) || + decoded.length !== 2 || + !Array.isArray(decoded[0]) || + decoded[1] === null || + typeof decoded[1] !== "object" || + Array.isArray(decoded[1]) + ) { + throw new SerializationError("CBOR call payload is not the [args, kwargs] wire shape"); + } + const [args, kwargs] = decoded as [unknown[], Record]; + return Object.keys(kwargs).length > 0 ? [...args, kwargs] : args; + } + + private body(bytes: Uint8Array): Uint8Array { + if (bytes.length === 0) { + throw new SerializationError("cannot deserialize an empty payload"); + } + if (bytes[0] === TAG_CBOR) { + return bytes.subarray(1); + } + if (bytes[0] === TAG_NATIVE) { + throw new SerializationError( + "payload is native-tagged (0x00): produced by a same-language-only serializer, not readable as CBOR wire format", + ); + } + throw new SerializationError( + `payload is not CBOR wire format (tag 0x${bytes[0]?.toString(16).padStart(2, "0")}, expected 0x02)`, + ); + } +} diff --git a/sdks/node/src/serializers/codec.ts b/sdks/node/src/serializers/codec.ts index 58c75ff3..50942b18 100644 --- a/sdks/node/src/serializers/codec.ts +++ b/sdks/node/src/serializers/codec.ts @@ -29,24 +29,44 @@ export class CodecSerializer implements Serializer { private readonly delegate: Serializer; private readonly codecs: readonly PayloadCodec[]; + /** Present only when the delegate is call-aware, so callers can detect it. */ + readonly serializeCall?: (args: unknown[]) => Uint8Array; + readonly deserializeCall?: (bytes: Uint8Array) => unknown[]; + constructor(delegate: Serializer = new JsonSerializer(), codecs: readonly PayloadCodec[] = []) { this.delegate = delegate; this.codecs = [...codecs]; + const serializeCall = delegate.serializeCall?.bind(delegate); + if (serializeCall) { + this.serializeCall = (args) => this.encode(serializeCall(args)); + } + const deserializeCall = delegate.deserializeCall?.bind(delegate); + if (deserializeCall) { + this.deserializeCall = (bytes) => deserializeCall(this.decode(bytes)); + } } serialize(value: unknown): Uint8Array { - let data = this.delegate.serialize(value); + return this.encode(this.delegate.serialize(value)); + } + + deserialize(bytes: Uint8Array): unknown { + return this.delegate.deserialize(this.decode(bytes)); + } + + private encode(data: Uint8Array): Uint8Array { + let out = data; for (const codec of this.codecs) { - data = codec.encode(data); + out = codec.encode(out); } - return data; + return out; } - deserialize(bytes: Uint8Array): unknown { - let data = bytes; + private decode(bytes: Uint8Array): Uint8Array { + let out = bytes; for (const codec of [...this.codecs].reverse()) { - data = codec.decode(data); + out = codec.decode(out); } - return this.delegate.deserialize(data); + return out; } } diff --git a/sdks/node/src/serializers/index.ts b/sdks/node/src/serializers/index.ts index 0955fac7..8623780f 100644 --- a/sdks/node/src/serializers/index.ts +++ b/sdks/node/src/serializers/index.ts @@ -1,9 +1,10 @@ export { AesGcmCodec } from "./aes-gcm"; +export { CborSerializer } from "./cbor"; export { CodecSerializer, type PayloadCodec } from "./codec"; export { EncryptedSerializer } from "./encrypted"; export { GzipCodec } from "./gzip"; export { HmacCodec } from "./hmac"; export { JsonSerializer } from "./json"; export { MsgpackSerializer } from "./msgpack"; -export type { Serializer } from "./serializer"; +export { deserializeCall, type Serializer, serializeCall } from "./serializer"; export { SignedSerializer } from "./signed"; diff --git a/sdks/node/src/serializers/serializer.ts b/sdks/node/src/serializers/serializer.ts index 937ede34..12c03df0 100644 --- a/sdks/node/src/serializers/serializer.ts +++ b/sdks/node/src/serializers/serializer.ts @@ -1,9 +1,29 @@ /** - * Pluggable codec for task arguments and results. Mirrors the Python shell's - * `Serializer` protocol so implementations can interoperate across languages. + * Pluggable codec for task arguments and results. Mirrors the cross-SDK + * serializer contract so implementations can interoperate across languages. * The Rust core stores payloads as opaque bytes. */ export interface Serializer { serialize(value: unknown): Uint8Array; deserialize(bytes: Uint8Array): unknown; + /** + * Optional call-shaped encoding: wire serializers write the cross-SDK call + * body `[args, kwargs]` (BINDING_CONTRACT.md) instead of a bare args array. + * Results always use `serialize`/`deserialize`. + */ + serializeCall?(args: unknown[]): Uint8Array; + /** Inverse of {@link serializeCall}: returns handler-ready positional args. */ + deserializeCall?(bytes: Uint8Array): unknown[]; +} + +/** Encode task args via `serializeCall` when the serializer is call-aware. */ +export function serializeCall(serializer: Serializer, args: unknown[]): Uint8Array { + return serializer.serializeCall ? serializer.serializeCall(args) : serializer.serialize(args); +} + +/** Decode a call payload into positional args, honoring call-aware serializers. */ +export function deserializeCall(serializer: Serializer, bytes: Uint8Array): unknown[] { + return serializer.deserializeCall + ? serializer.deserializeCall(bytes) + : (serializer.deserialize(bytes) as unknown[]); } diff --git a/sdks/node/src/worker.ts b/sdks/node/src/worker.ts index 6ac95d8d..44c34a46 100644 --- a/sdks/node/src/worker.ts +++ b/sdks/node/src/worker.ts @@ -19,7 +19,7 @@ import type { TaskConfigInput, } from "./native"; import { type ResourceRuntime, runWithResolver } from "./resources"; -import type { PayloadCodec, Serializer } from "./serializers"; +import { deserializeCall, type PayloadCodec, type Serializer } from "./serializers"; import type { QueueLimits, RegisteredTask, WorkerRunOptions } from "./types"; import { createLogger } from "./utils"; import type { WorkflowTracker } from "./workflows"; @@ -91,7 +91,7 @@ export class Worker { const taskCallback = async (invocation: JsTaskInvocation): Promise => { // Built-in workflow cache-return: echo the single (cached) arg as the result. if (invocation.taskName === CACHE_TASK) { - const [value] = serializer.deserialize(invocation.payload) as unknown[]; + const [value] = deserializeCall(serializer, invocation.payload); return Buffer.from(serializer.serialize(value)); } const task = tasks.get(invocation.taskName); @@ -107,7 +107,7 @@ export class Worker { } payload = codec.decode(payload); } - const args = serializer.deserialize(payload) as unknown[]; + const args = deserializeCall(serializer, payload); const ctx: TaskContext = { jobId: invocation.id, taskName: invocation.taskName, args }; // Resolve the middleware chain BEFORE allocating the cancel poller and // task scope — it reads storage and may throw, and nothing would clean diff --git a/sdks/node/src/workflows/manager.ts b/sdks/node/src/workflows/manager.ts index c0734c50..79defc27 100644 --- a/sdks/node/src/workflows/manager.ts +++ b/sdks/node/src/workflows/manager.ts @@ -1,6 +1,6 @@ import { WorkflowError } from "../errors"; import type { NativeQueue } from "../native"; -import type { Serializer } from "../serializers"; +import { type Serializer, serializeCall } from "../serializers"; import { WorkflowAnalysis, type WorkflowGraph } from "./analysis"; import { WorkflowBuilder } from "./builder"; import { WorkflowCacheStore } from "./cache"; @@ -47,7 +47,7 @@ export class WorkflowManager { private readonly encodePayload: (taskName: string, value: unknown) => Uint8Array = ( _taskName, value, - ) => this.serializer.serialize(value), + ) => serializeCall(this.serializer, value as unknown[]), ) { if (typeof this.native.submitWorkflow !== "function") { throw new WorkflowError("the native addon was built without the 'workflows' feature"); diff --git a/sdks/node/src/workflows/tracker.ts b/sdks/node/src/workflows/tracker.ts index 609a3e5f..fe084c49 100644 --- a/sdks/node/src/workflows/tracker.ts +++ b/sdks/node/src/workflows/tracker.ts @@ -1,6 +1,6 @@ import { createHash } from "node:crypto"; import type { JsOutcome, NativeQueue } from "../native"; -import type { Serializer } from "../serializers"; +import { type Serializer, serializeCall } from "../serializers"; import { createLogger } from "../utils"; import { CACHE_TASK, WorkflowCacheStore } from "./cache"; import { predecessorsOf, successorsOf, transitiveDeferred } from "./plan"; @@ -94,6 +94,10 @@ export class WorkflowTracker { constructor( private readonly native: NativeQueue, private readonly serializer: Serializer, + /** Per-task payload encoder (serializer + named codecs) so tracker-created + * jobs decode under the worker's unconditional codec reversal. */ + private readonly encodeCall: (taskName: string, args: unknown[]) => Uint8Array = (_, args) => + serializeCall(this.serializer, args), ) { this.cache = new WorkflowCacheStore(native); } @@ -310,7 +314,7 @@ export class WorkflowTracker { } const job = jobId ? this.native.getJob(jobId) : null; const result = job?.result ? this.serializer.deserialize(job.result) : null; - const payload = Buffer.from(this.serializer.serialize([result])); + const payload = Buffer.from(this.encodeCall(meta.compensate, [result])); this.native.enqueueCompensation( runId, node, @@ -394,7 +398,7 @@ export class WorkflowTracker { (JSON.parse(meta.fan_out).itemsFrom as string | null) ?? singlePred(plan, node); const items = this.readArrayResult(runId, itemsFrom); const childNames = items.map((_, i) => `${node}[${i}]`); - const childPayloads = items.map((item) => Buffer.from(this.serializer.serialize([item]))); + const childPayloads = items.map((item) => Buffer.from(this.encodeCall(meta.task_name, [item]))); this.native.expandFanOut( runId, @@ -434,7 +438,7 @@ export class WorkflowTracker { return; } // The combiner receives the whole list as its single positional argument. - const payload = Buffer.from(this.serializer.serialize([results])); + const payload = Buffer.from(this.encodeCall(meta.task_name, [results])); this.native.createDeferredJob( runId, fanInNode, @@ -585,7 +589,7 @@ export class WorkflowTracker { } const payload = meta.args_template ? Buffer.from(meta.args_template, "base64") - : Buffer.from(this.serializer.serialize([])); + : Buffer.from(this.encodeCall(meta.task_name, [])); this.native.createDeferredJob( runId, node, @@ -621,7 +625,8 @@ export class WorkflowTracker { return; } const value = this.serializer.deserialize(cached); - const payload = Buffer.from(this.serializer.serialize([value])); + // CACHE_TASK is internal and has no per-task codecs; keep the bare wire shape. + const payload = Buffer.from(serializeCall(this.serializer, [value])); this.native.createDeferredJob( runId, node, diff --git a/sdks/node/test/core/cbor-serializer.test.ts b/sdks/node/test/core/cbor-serializer.test.ts new file mode 100644 index 00000000..7b7e0960 --- /dev/null +++ b/sdks/node/test/core/cbor-serializer.test.ts @@ -0,0 +1,106 @@ +import { describe, expect, it } from "vitest"; +import { SerializationError } from "../../src/errors"; +import { CborSerializer } from "../../src/index"; +import { deserializeCall, JsonSerializer, serializeCall } from "../../src/serializers"; + +describe("CborSerializer", () => { + it("round-trips structured values", () => { + const s = new CborSerializer(); + const value = { a: 1, b: [2, 3], c: "x", d: true, e: null }; + expect(s.deserialize(s.serialize(value))).toEqual(value); + }); + + it("writes the 0x02 wire tag", () => { + const s = new CborSerializer(); + expect(s.serialize([1, 2])[0]).toBe(0x02); + expect(s.serializeCall([1, 2])[0]).toBe(0x02); + }); + + it("round-trips BigInt beyond Number.MAX_SAFE_INTEGER", () => { + const s = new CborSerializer(); + const big = 2n ** 64n + 1n; + expect(s.deserialize(s.serialize(big))).toBe(big); + }); + + it("round-trips Date values", () => { + const s = new CborSerializer(); + const moment = new Date("2026-07-11T12:30:45.000Z"); + expect(s.deserialize(s.serialize(moment))).toEqual(moment); + }); + + it("encodes calls as the [args, kwargs] wire shape", () => { + const s = new CborSerializer(); + const bytes = s.serializeCall([1, "a"]); + expect(s.deserialize(bytes)).toEqual([[1, "a"], {}]); + }); + + it("matches the BINDING_CONTRACT test vector for f(1, 'a')", () => { + const s = new CborSerializer(); + expect(Buffer.from(s.serializeCall([1, "a"])).toString("hex")).toBe("028282016161a0"); + }); + + it("deserializeCall returns positional args", () => { + const s = new CborSerializer(); + expect(s.deserializeCall(s.serializeCall([1, "a", { k: true }]))).toEqual([ + 1, + "a", + { k: true }, + ]); + }); + + it("surfaces non-empty kwargs as a trailing options object", () => { + const s = new CborSerializer(); + const encoder = s.serialize([["pos"], { named: 1 }]); + expect(s.deserializeCall(encoder)).toEqual(["pos", { named: 1 }]); + }); + + it("rejects a native-tagged (0x00) payload with a clear error", () => { + const s = new CborSerializer(); + expect(() => s.deserialize(new Uint8Array([0x00, 0x80]))).toThrow(/native-tagged/); + }); + + it("rejects untagged payloads", () => { + const s = new CborSerializer(); + expect(() => s.deserialize(new TextEncoder().encode('{"json":true}'))).toThrow( + SerializationError, + ); + }); + + it("rejects an empty payload", () => { + const s = new CborSerializer(); + expect(() => s.deserialize(new Uint8Array())).toThrow(/empty/); + }); + + it("rejects a call payload that is not [args, kwargs]", () => { + const s = new CborSerializer(); + expect(() => s.deserializeCall(s.serialize({ not: "a call" }))).toThrow(/wire shape/); + }); + + it("rejects call payloads with non-map kwargs", () => { + const s = new CborSerializer(); + for (const kwargs of [null, [], "oops", 7]) { + expect(() => s.deserializeCall(s.serialize([[1], kwargs]))).toThrow(/wire shape/); + } + }); + + it("encodes Map values as plain RFC 8949 maps, not tag 259", () => { + const s = new CborSerializer(); + const bytes = s.serialize(new Map([["k", 1]])); + // tag byte + a1 (1-entry map) + 61 6b ("k") + 01 — no d9 0103 (tag 259) prefix + expect(Buffer.from(bytes).toString("hex")).toBe("02a1616b01"); + }); +}); + +describe("call-shape helpers", () => { + it("fall back to plain serialize for non-call-aware serializers", () => { + const s = new JsonSerializer(); + const bytes = serializeCall(s, [1, 2]); + expect(s.deserialize(bytes)).toEqual([1, 2]); + expect(deserializeCall(s, bytes)).toEqual([1, 2]); + }); + + it("route through the wire shape for call-aware serializers", () => { + const s = new CborSerializer(); + expect(deserializeCall(s, serializeCall(s, [7]))).toEqual([7]); + }); +}); diff --git a/sdks/python/pyproject.toml b/sdks/python/pyproject.toml index 59482f81..6929d106 100644 --- a/sdks/python/pyproject.toml +++ b/sdks/python/pyproject.toml @@ -24,6 +24,7 @@ classifiers = [ "Topic :: System :: Distributed Computing", ] dependencies = [ + "cbor2>=5.6", "cloudpickle>=3.0", "msgpack>=1.0", 'tzdata; platform_system == "Windows"', diff --git a/sdks/python/taskito/__init__.py b/sdks/python/taskito/__init__.py index 88e1f002..614f05ff 100644 --- a/sdks/python/taskito/__init__.py +++ b/sdks/python/taskito/__init__.py @@ -48,6 +48,7 @@ from taskito.proxies.no_proxy import NoProxy from taskito.result import JobResult from taskito.serializers import ( + CborSerializer, CloudpickleSerializer, EncryptedSerializer, JsonSerializer, @@ -65,6 +66,7 @@ "BatchItemResult", "BatchPartialFailureError", "BatchResultTypeError", + "CborSerializer", "CircuitBreakerOpenError", "CircularDependencyError", "CloudpickleSerializer", diff --git a/sdks/python/taskito/serializers.py b/sdks/python/taskito/serializers.py index f97cb2e8..cfc17e2a 100644 --- a/sdks/python/taskito/serializers.py +++ b/sdks/python/taskito/serializers.py @@ -7,14 +7,18 @@ import json from typing import Any, Protocol, runtime_checkable +import cbor2 import cloudpickle import msgpack -# Format tags for the envelope written by ``SmartSerializer``. A legacy +# Wire-envelope tags: the first payload byte records which codec produced the +# body. Tags 0x00-0x01 predate the cross-SDK contract; 0x02+ are part of it +# (see crates/taskito-core/BINDING_CONTRACT.md "Wire envelope"). A legacy # cloudpickle payload starts with the pickle protocol-2+ opcode ``\x80``, which # never collides with these tags — so untagged bytes are unambiguously legacy. -_CODEC_CLOUDPICKLE = b"\x00" +_CODEC_CLOUDPICKLE = b"\x00" # native: same-language only, never cross-SDK _CODEC_MSGPACK = b"\x01" +_CODEC_CBOR = b"\x02" # cross-SDK default wire format # msgpack has no native tuple type and would silently flatten tuples to lists. # A custom ExtType preserves them so payloads round-trip with exact Python @@ -103,6 +107,38 @@ def loads(self, data: bytes) -> Any: return msgpack.unpackb(data, raw=False) +class CborSerializer: + """CBOR-based serializer for cross-SDK payloads (RFC 8949). + + Writes the ``0x02`` wire-envelope tag followed by a CBOR body, per the + cross-SDK contract in ``BINDING_CONTRACT.md``. Use for tasks produced or + consumed by another SDK: unlike JSON, CBOR round-trips big integers + (bignum), ``datetime`` (tag 0), ``bytes``, and ``Decimal`` (tag 4) + losslessly across languages. + + Only CBOR-representable types are supported. Tuples become arrays on the + wire (like JSON/msgpack); naive ``datetime`` values are rejected by cbor2 — + use timezone-aware datetimes. For arbitrary Python objects keep the default + :class:`SmartSerializer` (same-language only). + """ + + def dumps(self, obj: Any) -> bytes: + return _CODEC_CBOR + cbor2.dumps(obj) + + def loads(self, data: bytes) -> Any: + if not data: + raise ValueError("Cannot deserialize empty payload") + tag, body = data[:1], data[1:] + if tag == _CODEC_CBOR: + return cbor2.loads(body) + if tag == _CODEC_CLOUDPICKLE: + raise ValueError( + "Payload is native-tagged (0x00): produced by a same-language-only " + "serializer, not readable as CBOR wire format" + ) + raise ValueError(f"Payload is not CBOR wire format (tag {data[0]:#04x}, expected 0x02)") + + class SmartSerializer: """Default serializer: msgpack for plain payloads, cloudpickle fallback. @@ -134,6 +170,10 @@ def loads(self, data: bytes) -> Any: return _msgpack_unpackb(body) if tag == _CODEC_CLOUDPICKLE: return cloudpickle.loads(body) + if tag == _CODEC_CBOR: + # Cross-SDK wire payload (e.g. enqueued by another SDK's + # CborSerializer) — readable without per-task configuration. + return cbor2.loads(body) # Untagged: a legacy cloudpickle payload from before the envelope existed. return cloudpickle.loads(data) diff --git a/sdks/python/tests/core/test_serializers.py b/sdks/python/tests/core/test_serializers.py index 88fa786d..6992d511 100644 --- a/sdks/python/tests/core/test_serializers.py +++ b/sdks/python/tests/core/test_serializers.py @@ -6,6 +6,7 @@ import pytest from taskito.serializers import ( + CborSerializer, CloudpickleSerializer, JsonSerializer, Serializer, @@ -219,6 +220,66 @@ def test_empty_payload_rejected(self) -> None: def test_satisfies_protocol(self) -> None: assert isinstance(SmartSerializer(), Serializer) + def test_loads_cbor_wire_payload(self) -> None: + """Cross-SDK CBOR payloads (tag 0x02) load without per-task config.""" + encoded = CborSerializer().dumps({"from": "another-sdk", "n": 2**80}) + assert SmartSerializer().loads(encoded) == {"from": "another-sdk", "n": 2**80} + + +class TestCborSerializer: + def test_roundtrip_dict(self) -> None: + s = CborSerializer() + data = {"key": "value", "num": 42, "nested": [1, 2, 3], "raw": b"\x00\xff"} + assert s.loads(s.dumps(data)) == data + + def test_wire_tag_is_0x02(self) -> None: + assert CborSerializer().dumps([1, 2])[:1] == b"\x02" + + def test_big_int_roundtrip(self) -> None: + """Integers beyond 2^53 (JS Number limit) and 2^64 (CBOR bignum) + must survive — the reason CBOR is the cross-SDK default over JSON.""" + s = CborSerializer() + for val in [2**53 + 1, 2**64 + 1, -(2**80)]: + assert s.loads(s.dumps(val)) == val + + def test_datetime_roundtrip(self) -> None: + from datetime import datetime, timezone + + s = CborSerializer() + moment = datetime(2026, 7, 11, 12, 30, 45, tzinfo=timezone.utc) + assert s.loads(s.dumps(moment)) == moment + + def test_call_envelope_shape(self) -> None: + """An ``(args, kwargs)`` payload becomes a 2-element CBOR array — + the cross-SDK call-body shape from BINDING_CONTRACT.md.""" + s = CborSerializer() + restored = s.loads(s.dumps(((1, "a"), {"k": True}))) + assert restored == [[1, "a"], {"k": True}] + + def test_matches_binding_contract_vector(self) -> None: + """Byte-exact against the BINDING_CONTRACT.md test vector for + call ``f(1, "a")`` — the fixture every SDK's tests assert.""" + s = CborSerializer() + assert s.dumps(((1, "a"), {})).hex() == "028282016161a0" + assert s.loads(bytes.fromhex("028282016161a0")) == [[1, "a"], {}] + + def test_rejects_native_tagged_payload(self) -> None: + native = SmartSerializer().dumps(lambda x: x) + assert native[:1] == b"\x00" + with pytest.raises(ValueError, match="native-tagged"): + CborSerializer().loads(native) + + def test_rejects_untagged_payload(self) -> None: + with pytest.raises(ValueError, match="not CBOR wire format"): + CborSerializer().loads(b'{"json": true}') + + def test_empty_payload_rejected(self) -> None: + with pytest.raises(ValueError, match="empty payload"): + CborSerializer().loads(b"") + + def test_satisfies_protocol(self) -> None: + assert isinstance(CborSerializer(), Serializer) + class TestSignedSerializer: def test_roundtrip(self) -> None: