Pluggable Serializers
Built-in JSON, MessagePack, CBOR, signed, and encrypted serializers, plus the cross-SDK payload-codec chain for compression and encryption.
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.
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.
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.
from flexiq import Queue
queue = Queue() # uses SmartSerializerHandles 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.
from flexiq import CloudpickleSerializer, Queue
queue = Queue(serializer=CloudpickleSerializer())Human-readable JSON payloads. Useful for debugging, cross-language interop, or when arguments and results are plain types.
from flexiq import JsonSerializer, Queue
queue = Queue(serializer=JsonSerializer())import { Queue, JsonSerializer } from "@byteveda/flexiq";
new Queue({ dbPath: "flexiq.db", serializer: new JsonSerializer() });import org.byteveda.flexiq.FlexiQ;
import org.byteveda.flexiq.serialization.JsonSerializer;
FlexiQ flexiq = FlexiQ.builder()
.sqlite("flexiq.db")
.serializer(new JsonSerializer())
.open();Compact binary serialization — smaller payloads than JSON, cross-language compatible.
from flexiq import MsgPackSerializer, Queue
queue = Queue(serializer=MsgPackSerializer())import { Queue, MsgpackSerializer } from "@byteveda/flexiq";
new Queue({ dbPath: "flexiq.db", serializer: new MsgpackSerializer() });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.
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.
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.
from flexiq import CborSerializer, Queue
queue = Queue(serializer=CborSerializer())import { Queue, CborSerializer } from "@byteveda/flexiq";
new Queue({ dbPath: "flexiq.db", serializer: new CborSerializer() });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.
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.
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.
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!),
});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();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.
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.
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!),
});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:
python -c "import os, base64; print(base64.b64encode(os.urandom(32)).decode())"For both confidentiality and integrity, wrap one in the other:
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.
Implement the Serializer interface with two methods:
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 });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.
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.
| 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.
| 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.
| 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.
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.
Apply a chain queue-wide — it wraps the queue serializer, so it covers every payload and result:
from flexiq import AesGcmCodec, GzipCodec, Queue
queue = Queue(codec=[GzipCodec(), AesGcmCodec(key)]) # compress, then encryptimport { Queue, GzipCodec, AesGcmCodec } from "@byteveda/flexiq";
new Queue({
dbPath: "flexiq.db",
codec: [new GzipCodec(), new AesGcmCodec(key)], // compress, then encrypt
});FlexiQ flexiq = FlexiQ.builder()
.sqlite("flexiq.db")
.codec(new GzipCodec(), new AesGcmCodec(key)) // compress, then encrypt
.open();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.
Register codecs under a name and opt individual tasks in — payload only, results still use the plain queue serializer:
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"] },
);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 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.