Node/Guides/Extensibility/

Pluggable Serializers

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 Rust core stores every payload as an opaque byte blob, so serialization is entirely a host-language concern — producers and workers just need to agree on the same serializer.

Producers and workers must match

The serializer is not negotiated at runtime — it's configured once per Queue. A job enqueued with one serializer can only be decoded by a worker configured with a compatible one.

The default is SmartSerializer — MessagePack for plain data (dicts, lists, strings, numbers, booleans, None, tuples), with an automatic CloudpickleSerializer fallback for anything MessagePack can't encode (lambdas, closures, custom class instances). No configuration needed.

The default is JsonSerializer — human-readable and easy to debug. Other built-ins: MsgpackSerializer, CborSerializer, SignedSerializer, EncryptedSerializer.

The default is JsonSerializer — Jackson-backed, human-readable and easy to debug. Other built-ins: MsgpackSerializer, CborSerializer, SignedSerializer, EncryptedSerializer.

Built-in serializers

SmartSerializer (default)

Encodes plain data via MessagePack — fast and compact — and transparently falls back to CloudpickleSerializer for anything MessagePack can't encode. A one-byte tag on each payload records which codec produced it, so loads() always picks the right path.

python

from flexiq import Queue

queue = Queue()  # uses SmartSerializer

CloudpickleSerializer

Handles lambdas, closures, and complex Python objects unconditionally — every payload goes through cloudpickle, skipping SmartSerializer's MessagePack fast path. Use it directly for that predictability, or as the inner serializer for EncryptedSerializer / SignedSerializer.

python

from flexiq import CloudpickleSerializer, Queue

queue = Queue(serializer=CloudpickleSerializer())

JsonSerializer

Human-readable JSON payloads. Useful for debugging, cross-language interop, or when arguments and results are plain types.

python

from flexiq import JsonSerializer, Queue

queue = Queue(serializer=JsonSerializer())

import { Queue, JsonSerializer } from "@byteveda/flexiq";

new Queue({ dbPath: "flexiq.db", serializer: new JsonSerializer() });

java

import org.byteveda.flexiq.FlexiQ;
import org.byteveda.flexiq.serialization.JsonSerializer;

FlexiQ flexiq = FlexiQ.builder()
        .sqlite("flexiq.db")
        .serializer(new JsonSerializer())
        .open();

MessagePack

Compact binary serialization — smaller payloads than JSON, cross-language compatible.

python

from flexiq import MsgPackSerializer, Queue

queue = Queue(serializer=MsgPackSerializer())

import { Queue, MsgpackSerializer } from "@byteveda/flexiq";

new Queue({ dbPath: "flexiq.db", serializer: new MsgpackSerializer() });

java

import org.byteveda.flexiq.FlexiQ;
import org.byteveda.flexiq.serialization.MsgpackSerializer;

FlexiQ flexiq = FlexiQ.builder()
        .sqlite("flexiq.db")
        .serializer(new MsgpackSerializer())
        .open();

msgpack ships as a core flexiq dependency, so no extra install is required — pip install flexiq[msgpack] exists only for discoverability if you want to spell it out in your own requirements file.

Type restrictions

MsgPackSerializer only handles basic types: dicts, lists, strings, numbers, booleans, and None. It does not support lambdas, closures, or arbitrary Python objects. Use CloudpickleSerializer when you need to pass complex objects.

@msgpack/msgpack ships as a core dependency — no extra install required.

MsgpackSerializer's org.msgpack:jackson-dataformat-msgpack dependency is compileOnly — add it to your build explicitly.

CborSerializer

Binary serialization writing the 0x02 wire-envelope tag — the default format for tasks produced or consumed by another FlexiQ SDK. Unlike JSON, big integers, datetime/Date, bytes, and decimals round-trip losslessly across languages.

python

from flexiq import CborSerializer, Queue

queue = Queue(serializer=CborSerializer())

import { Queue, CborSerializer } from "@byteveda/flexiq";

new Queue({ dbPath: "flexiq.db", serializer: new CborSerializer() });

java

import org.byteveda.flexiq.FlexiQ;
import org.byteveda.flexiq.serialization.CborSerializer;

FlexiQ flexiq = FlexiQ.builder()
        .sqlite("flexiq.db")
        .serializer(new CborSerializer())
        .open();

cbor2 ships as a required flexiq dependency — no extra install needed. The default SmartSerializer can already read 0x02-tagged CBOR payloads (for example one written by another FlexiQ 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.

Cross-SDK payload contract

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 the payload — it authenticates it. A worker refuses to deserialize bytes that were not produced with the shared key, so an attacker who can write to the queue's storage cannot smuggle in a forged payload.

python

import os
from flexiq import Queue, SignedSerializer, SmartSerializer

key = os.urandom(32)  # share this across producers and workers
queue = Queue(serializer=SignedSerializer(SmartSerializer(), key))

import { Queue, SignedSerializer } from "@byteveda/flexiq";

new Queue({
  dbPath: "flexiq.db",
  serializer: new SignedSerializer(process.env.FLEXIQ_SECRET!),
});

java

import java.nio.charset.StandardCharsets;
import org.byteveda.flexiq.FlexiQ;
import org.byteveda.flexiq.serialization.JsonSerializer;
import org.byteveda.flexiq.serialization.SignedSerializer;

byte[] key = System.getenv("FLEXIQ_SIGNING_KEY").getBytes(StandardCharsets.UTF_8);

FlexiQ flexiq = FlexiQ.builder()
        .sqlite("flexiq.db")
        .serializer(new SignedSerializer(new JsonSerializer(), key))
        .open();

Why this matters

The default serializer can execute code on load (cloudpickle handles lambdas and arbitrary objects). Without signing, anyone able to write to the backing store can achieve remote code execution on every worker that dequeues a crafted job. SignedSerializer closes that path. The key must be at least 32 bytes of CSPRNG output and identical on producers and workers.

The key is a plain secret string, hashed internally — producers and workers just need to share the same string.

EncryptedSerializer

AES-256-GCM encryption. Payloads stored in the database are opaque ciphertext — only the key holder can read them. Provides confidentiality and integrity, a superset of SignedSerializer.

python

import base64
import os
from flexiq import CloudpickleSerializer, EncryptedSerializer, Queue

key = base64.b64decode(os.environ["QUEUE_KEY"])  # 16, 24, or 32 raw bytes
queue = Queue(serializer=EncryptedSerializer(CloudpickleSerializer(), key))

import { Queue, EncryptedSerializer } from "@byteveda/flexiq";

new Queue({
  dbPath: "flexiq.db",
  serializer: new EncryptedSerializer(process.env.FLEXIQ_SECRET!),
});

java

import java.util.Base64;
import org.byteveda.flexiq.FlexiQ;
import org.byteveda.flexiq.serialization.JsonSerializer;
import org.byteveda.flexiq.serialization.EncryptedSerializer;

byte[] key = Base64.getDecoder().decode(System.getenv("FLEXIQ_ENC_KEY")); // 16, 24, or 32 bytes

FlexiQ flexiq = FlexiQ.builder()
        .sqlite("flexiq.db")
        .serializer(new EncryptedSerializer(new JsonSerializer(), key))
        .open();

EncryptedSerializer always wraps an explicit inner serializer — there's no implicit default. The key must be exactly 16, 24, or 32 bytes (AES-128/192/256), base64-decoded if it came from a string. Generate one with:

bash

python -c "import os, base64; print(base64.b64encode(os.urandom(32)).decode())"

For both confidentiality and integrity, wrap one in the other:

python

from flexiq import EncryptedSerializer, Queue, SignedSerializer, SmartSerializer

inner = EncryptedSerializer(SmartSerializer(), key=enc_key)
queue = Queue(serializer=SignedSerializer(inner, sign_key))

The key is derived from secret via SHA-256, so producers and workers only need to share the string — not raw key bytes.

Custom serializers

Implement the Serializer interface with two methods:

python

import msgpack
from flexiq import Queue, Serializer


class MyMsgpackSerializer:
    def dumps(self, obj) -> bytes:
        return msgpack.packb(obj)

    def loads(self, data: bytes):
        return msgpack.unpackb(data, raw=False)


queue = Queue(serializer=MyMsgpackSerializer())

import { Queue } from "@byteveda/flexiq";
import type { Serializer } from "@byteveda/flexiq";

const mySerializer: Serializer = {
  serialize: (value) => Buffer.from(JSON.stringify(value)),
  deserialize: (bytes) => JSON.parse(Buffer.from(bytes).toString()),
};

new Queue({ dbPath: "flexiq.db", serializer: mySerializer });

java

import org.byteveda.flexiq.serialization.Serializer;

public final class MySerializer implements Serializer {
    @Override
    public byte[] serialize(Object value) { /* ... */ }

    @Override
    public <T> T deserialize(byte[] bytes, Class<T> type) { /* ... */ }
}

Method	Signature	Description
`dumps`	`(obj: Any) -> bytes`	Serialize an object to bytes
`loads`	`(data: bytes) -> Any`	Deserialize bytes back to an object

The serializer is used for both task arguments (the (args, kwargs) tuple) and return values.

Result deserialization

job.result() uses the queue's configured serializer for deserialization. If you're using JsonSerializer or a custom serializer, results are correctly deserialized with that serializer — not hardcoded cloudpickle.

The Type overload (deserialize(byte[] bytes, Type type)) is a default method supporting generic payloads from a TypeReference; the base implementation handles plain Class types and rejects generic ones — override it in a generics-aware implementation, as JsonSerializer does.

Choosing a serializer

	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` (`flexiq[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 CborSerializer when a task is produced or consumed by another FlexiQ 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 for a shorter overview and what gets serialized where.

Choosing a serializer

Serializer	When to use	Cross-SDK?
`JsonSerializer`	The default — human-readable, debuggable payloads for Node↔Node work. Leave it unless you have a reason to switch.	No — portable format, same-SDK payloads
`MsgpackSerializer`	Node↔Node work where you want compact binary payloads instead of JSON text.	No — portable format, same-SDK payloads
`CborSerializer`	A task is produced or consumed by another FlexiQ SDK. The only serializer whose wire format (the `0x02` envelope tag) is part of the cross-SDK contract; round-trips big integers as `BigInt`.	Yes — cross-SDK wire format
`SignedSerializer(secret, inner?)`	Payloads must be tamper-evident but not hidden — prefixes an HMAC-SHA256 tag and rejects mismatches. Authentication, not encryption.	Same-SDK by default (wraps `JsonSerializer`)
`EncryptedSerializer(secret, inner?)`	Tasks carry sensitive data that must not be readable in the database — AES-256-GCM confidentiality and integrity.	Same-SDK by default (wraps `JsonSerializer`)

Rule of thumb: keep the default JsonSerializer unless you have a specific reason to switch. Use MsgpackSerializer for more compact Node↔Node payloads, CborSerializer when a task crosses SDK boundaries, and EncryptedSerializer when payloads must not be readable in storage.

Choosing a serializer

Serializer	When to use	Cross-SDK?
`JsonSerializer`	The default — JSON via Jackson, accepts a custom `ObjectMapper`. Human-readable and debuggable for Java↔Java work.	No — portable format, same-SDK payloads
`MsgpackSerializer`	Java↔Java work where you want compact binary payloads. Its `jackson-dataformat-msgpack` dependency is `compileOnly` — add it to your build.	No — portable format, same-SDK payloads
`CborSerializer`	A task is produced or consumed by another FlexiQ SDK. The only serializer whose wire format (the `0x02` envelope tag) is part of the cross-SDK contract. `jackson-dataformat-cbor` is `compileOnly`.	Yes — cross-SDK wire format
`SignedSerializer(delegate, key)`	Payloads must be tamper-evident but not hidden — prefixes an HMAC-SHA256 tag with constant-time verification. Authentication, not encryption.	Same-SDK by default (depends on `delegate`)
`EncryptedSerializer(delegate, key)`	Tasks carry sensitive data that must not be readable in the database — AES-GCM confidentiality and integrity. Key must be 16, 24, or 32 bytes.	Same-SDK by default (depends on `delegate`)

Rule of thumb: keep the default JsonSerializer unless you have a specific reason to switch. Use MsgpackSerializer for more compact Java↔Java payloads, CborSerializer when a task crosses SDK boundaries, and EncryptedSerializer when payloads must not be readable in storage.

Payload codecs

A PayloadCodec is a reversible byte-to-byte transform layered around a serializer — compression, encryption, signing — applied after serialization on the producer and reversed before deserialization on the worker. Wire formats are part of the cross-SDK wire contract, so a codec-framed payload decodes from any FlexiQ binding:

Codec	Wire format	Description
`GzipCodec`	standard gzip stream	Compression; decompression capped at 64 MiB by default (zip-bomb guard)
`AesGcmCodec(key)`	`[12-byte nonce][ciphertext \|\| 16-byte GCM tag]`	AES-GCM encryption, fresh nonce per payload. Key must be 16, 24, or 32 bytes
`HmacCodec(key)`	`[32-byte mac][body]`	HMAC-SHA256 signing, constant-time verification. Key must not be empty

A chain encodes in list order on the producer and decodes in reverse on the worker — put GzipCodec before a signing or encryption codec so integrity is verified before decompressing.

Global codec chain

Apply a chain queue-wide — it wraps the queue serializer, so it covers every payload and result:

python

from flexiq import AesGcmCodec, GzipCodec, Queue

queue = Queue(codec=[GzipCodec(), AesGcmCodec(key)])  # compress, then encrypt

import { Queue, GzipCodec, AesGcmCodec } from "@byteveda/flexiq";

new Queue({
  dbPath: "flexiq.db",
  codec: [new GzipCodec(), new AesGcmCodec(key)], // compress, then encrypt
});

java

FlexiQ flexiq = FlexiQ.builder()
        .sqlite("flexiq.db")
        .codec(new GzipCodec(), new AesGcmCodec(key)) // compress, then encrypt
        .open();

Jobs persisted before the chain was enabled

Jobs written before a codec chain is turned on cannot be decoded through it — drain the queue before turning codecs on, or before changing the chain.

Named codecs (per task)

python

from flexiq import AesGcmCodec, Queue

queue = Queue(codecs={"secret": AesGcmCodec(key)})

@queue.task(codecs=["secret"])
def export_report(report_id: int):
    ...

import { Queue, AesGcmCodec } from "@byteveda/flexiq";

const queue = new Queue({
  dbPath: "flexiq.db",
  codecs: { secret: new AesGcmCodec(key) },
});

queue.task(
  "exportReport",
  (reportId: number) => {
    // ...
  },
  { codecs: ["secret"] },
);

java

FlexiQ flexiq = FlexiQ.builder()
        .sqlite("flexiq.db")
        .codec("secret", new AesGcmCodec(key))
        .open();

Task<Report> export = Task.of("export", Report.class).codecs("secret");

Handlers annotated @Compressed / @Encrypted get the "compressed" / "encrypted" codec names recorded on their generated task constants — register codecs under those names on both producers and workers. The key is supplied at runtime, never in the annotation.

The same chain (or the same names) must be configured on producers and workers. Dedup keys for auto-derived idempotency hash the pre-codec payload, so a nondeterministic named codec (an AES-GCM nonce, for example) never breaks dedup — see Idempotency for the full precedence rules and the hash recipe.

The global chain is different from named codecs

The warning above is about per-task codecs=[...]. The queue-wide codec= chain wraps the serializer itself, so a nondeterministic codec there (e.g. AesGcmCodec) does make idempotent=True auto-keys nondeterministic from one call to the next — pass an explicit idempotency_key instead when a global codec chain is enabled.