Predicates

The predicate receives the task name, the (already onEnqueue-rewritten) args typed to the task's signature, and now — the wall clock read once for this evaluation, so a time-based gate can be unit-tested against a pinned instant instead of the real clock.

The PredicateContext carries the task name and the payload — after onEnqueue middleware has run, so gates see the rewritten payload.

What predicates do not do

Predicates are not the right place to rebuild flexiq's atomic enforcement primitives. Use the dedicated knobs:

Each is enforced atomically in the Rust scheduler. A Python predicate that races against them is weaker and harder to reason about. Predicates are for gates the scheduler doesn't already provide: time windows, payload-driven branching, feature flags, environment config, and custom Python logic.

What replaces a predicate

With nothing to register, the question becomes where each kind of check belongs. Most of them were never a gate's job: the scheduler enforces them atomically already, and a #[flexiq::task] attribute is how you ask for one.

What is left over — time windows, feature flags, tenant allowlists, payload-driven branching — is ordinary Rust. The only thing a registered gate would have added is a place to hang it.

@queue.task accepts exactly one predicate= per task — compose everything you need with & / | / ~ before passing it in. A & B short-circuits if A denies; A | B tries both and, if neither allows, surfaces the more informative outcome; ~A inverts a boolean result (see Richer outcomes for how Defer / Cancel behave under composition).

Registering more than one gate on the same task also composes them — they all must pass:

queue.gate("charge", ({ args }) => args[0] > 0);
queue.gate("charge", ({ args }) => args[0] <= 10_000);

Registering more than one predicate() (or gate()) on the same task also composes them — they run in registration order and the first non-allow decision wins.

Nothing is registered, so there is no second registration to combine with. Two checks are two ifs, or one function returning bool that the enqueue site and its unit test both call.

evaluate() can also return Defer or Cancel instead of a plain boolean:

A predicate that raises is treated as False (fail-closed). Errors are logged and counted on PredicateMetrics.

Composition respects these richer outcomes: A & B short-circuits on the first Defer/Cancel/False from A; A | B evaluates both and, if neither allows, returns the most informative denial (Cancel > Defer > False); ~A inverts True/False but passes Defer and Cancel through unchanged — they're terminal outcomes, not booleans.

from typing import Any
from flexiq import Queue
from flexiq.predicates import Predicate, PredicateContext

queue = Queue()

@queue.register_predicate("tenant_quota_under")
class TenantQuotaUnder(Predicate):
    OP = "tenant_quota_under"

    def __init__(self, limit: int = 0) -> None:
        self.limit = limit

    def evaluate(self, ctx: PredicateContext) -> bool:
        tenant = ctx.kwargs.get("tenant")
        return tenant is not None and quota_service.used(tenant) < self.limit

    def to_dict(self) -> dict[str, Any]:
        return {"op": "tenant_quota_under", "limit": self.limit}

    @classmethod
    def _from_kwargs(cls, kwargs: dict[str, Any]) -> "Predicate":
        return cls(**kwargs)

Once registered with queue.register_predicate(op), a new op is usable in JSON, the string DSL, and operator composition exactly like a built-in recipe (see The DSL surface). Predicates may also be async — return a coroutine from evaluate and it will be awaited transparently.

A gate may return a Decision instead of a boolean:

import { Decision } from "@byteveda/flexiq";

queue.gate("digest", ({ args }) => {
  if (!tenants.has(args[0] as string)) return Decision.reject("unknown tenant");
  if (!quota.available(args[0] as string)) return Decision.defer(5 * 60_000);
  return Decision.allow();
});

queue.enqueue("digest", ["acme"]); // throws if the gate rejects or skips
queue.tryEnqueue("digest", ["acme"]); // null instead of throwing on a skip

Gates run in registration order and the first non-allow decision wins, so a gate that defers short-circuits the ones after it. A gate that throws is a bug rather than a denial: the error propagates to the caller (and increments errors on predicateStats()).

Gates are evaluated only at enqueue time. To react once a job is already running, use middleware or conditions inside a workflow.

gate(taskName, g) registers an EnqueueGate, whose EnqueueDecision can do more than pass/fail. Gates run in registration order; the first non-allow decision wins.

flexiq.gate("charge", ctx -> switch (classify(ctx.payload())) {
    case OK        -> EnqueueDecision.allow();
    case DUPLICATE -> EnqueueDecision.skip("already charged");
    case TOO_EARLY -> EnqueueDecision.defer(Duration.ofHours(1));
    case FRAUD     -> EnqueueDecision.reject("fraud check failed");
});

tryEnqueue is the gate-aware form — a skip becomes an empty Optional instead of an exception:

Optional<String> id = flexiq.tryEnqueue(charge, amount);

predicate(taskName, p) is sugar for a gate that only ever allows or rejects — use it for plain boolean checks and reach for gate() when you need skip/defer too.

There is no decision type to return, but each of the four shapes it names is already something you can do:

The producer-side column is the one to prefer. A refusal there costs a branch; a refusal at dispatch has already cost a row, a claim and a dispatch.

All recipes are registered AST ops. Import from flexiq.predicates.

Time

Payload

Defensive

External config

Pluggable feature-flag provider

from flexiq.predicates import FeatureFlagProvider, register_feature_flag_provider

class LaunchDarklyProvider:
    def __init__(self, client) -> None:
        self._c = client

    def is_enabled(self, name, ctx) -> bool:
        user = {"key": ctx.kwargs.get("tenant_id", "anon")}
        return bool(self._c.variation(name, user, False))

register_feature_flag_provider("launchdarkly", LaunchDarklyProvider(ldclient.get()))

# Now both Python and string forms round-trip:
p = feature_flag("billing", provider="launchdarkly")
assert Predicate.from_dict(p.to_dict()).to_dict() == p.to_dict()

Recipes ships ready-made gates for common policies — time-based ones defer out-of-window enqueues to the next open moment; filters skip silently:

flexiq.gate("dispatch", Recipes.businessHours(ZoneId.of("America/New_York")));
flexiq.gate("digest", Recipes.dayOfWeek(zone, DayOfWeek.MONDAY, DayOfWeek.THURSDAY));
flexiq.gate("sync", Recipes.timeWindow(zone, LocalTime.of(22, 0), LocalTime.of(6, 0)));
flexiq.gate("beta", Recipes.featureFlag("beta-export", flags));
flexiq.gate("charge", Recipes.payloadMatches(p -> ((Order) p).amount() > 0));

Import the recipes individually, or as the Recipes bundle. Each one's outcome is listed below and they are not uniform — the window recipes (businessHours, timeWindow, dayOfWeek) defer to the next open moment, payloadMatches and featureFlag skip silently, and isWeekend / before / envVarTruthy are plain booleans, so on their own a false rejects like any other boolean gate. Compose those three with not / allOf / anyOf rather than registering them bare if a rejection isn't what you want.

Every argument is validated when the recipe is built, so a bad zone or window throws PredicateValidationError at wiring time rather than on the first enqueue.

Time

Zones are IANA names resolved through Intl (no extra dependency) and default to "UTC". A target across a DST boundary resolves to the correct instant.

Payload

External config

import { Recipes } from "@byteveda/flexiq";

const zone = "America/New_York";
queue.gate("dispatch", Recipes.businessHours({ timeZone: zone }));
queue.gate("digest", Recipes.dayOfWeek({ days: ["mon", "thu"], timeZone: zone }));
queue.gate("sync", Recipes.timeWindow({ start: "22:00", end: "06:00", timeZone: zone }));
queue.gate("beta", Recipes.featureFlag("beta-export"));
queue.gate("charge", Recipes.payloadMatches("args.0.currency", "USD"));

Pluggable feature-flag provider

featureFlag takes either a FeatureFlagProvider or a plain lookup function:

import { featureFlag, type FeatureFlagProvider } from "@byteveda/flexiq";

const launchDarkly: FeatureFlagProvider = {
  isEnabled(flag, ctx) {
    return client.variation(flag, { key: String(ctx.args[0] ?? "anon") }, false);
  },
};

queue.gate("billing", featureFlag("billing", launchDarkly));
queue.gate("beta", featureFlag("beta", (flag) => myFlags[flag] === true));

Naming a gate

Register a gate by name to drive gating from configuration — the name is the portable part, since a Node gate is a closure with no serializable form:

import { defaultRegistry, registerPredicate, Recipes } from "@byteveda/flexiq";

registerPredicate("business-hours", Recipes.businessHours({ timeZone: zone }));

queue.gate("dispatch", process.env.DISPATCH_GATE ?? "business-hours");
defaultRegistry().names(); // ["business-hours"]

The name is resolved when gate() is called, so an unknown one throws PredicateValidationError at startup rather than on the first enqueue.

There is no recipe library — nothing here ships a business-hours, day-of-week, time-window, payload-match or feature-flag helper, and there is no named registry to look one up in. Each is a few lines of your own Rust, and being your own code is what lets a time-based check take now as an argument and be unit-tested against a pinned instant instead of the real clock.

The DSL surface

Every predicate is a node in a closed, registered AST. Three equivalent authoring paths produce the same tree — Python operators (above), JSON, and a small string DSL:

from flexiq.predicates import Predicate

p = Predicate.from_dict({
    "op": "or",
    "args": [
        {"op": "and", "args": [
            {"op": "is_business_hours", "tz": "US/Pacific"},
            {"op": "not", "arg": {"op": "queue_paused"}},
        ]},
        {"op": "feature_flag", "flag": "new", "provider": "env"},
    ],
})

from flexiq.predicates import parse

p = parse(
    'is_business_hours(tz="US/Pacific") & !queue_paused() '
    '| feature_flag(flag="new", provider="env")'
)

The string DSL supports & / | / ! and the long forms and / or / not. Literals: strings, ints, floats, true / false / null, and [lists, of, literals]. Round-trip is stable:

from flexiq.predicates import format_predicate, parse

p1 = is_business_hours() & ~queue_paused()
s = format_predicate(p1)        # 'is_business_hours() & !queue_paused()'
p2 = parse(s)
assert p1.to_dict() == p2.to_dict()

The string form is the right surface for ops-style editing — storing the gate in a config file, or letting an operator tweak it through a dashboard textbox without a code deploy:

@queue.task(predicate=parse(
    'is_business_hours(tz="US/Pacific") & !queue_paused()'
))
def send_report(): ...

Where predicates run

A predicate registered on a task is evaluated twice:

1. Enqueue time — inside Queue.enqueue() / enqueue_many(). Defer adjusts the caller's delay=. Cancel (or False with on_false="cancel") raises PredicateRejectedError and the job is never saved.
2. Worker-dispatch time — just before the task body runs. Cancel raises TaskCancelledError (the job is marked cancelled). Defer re-enqueues a fresh job with the same payload and delay; the current job is cancelled.

PredicateContext.job_id is None at enqueue time, set at dispatch time — useful for custom predicates that need to behave differently across phases.

Middleware filtering

TaskMiddleware accepts predicate= to gate which jobs the middleware applies to. The legacy contrib task_filter=Callable[[str], bool] kwarg still works and is translated internally to a predicate.

from flexiq.contrib.sentry import SentryMiddleware
from flexiq.predicates import payload_matches

queue = Queue(middleware=[
    SentryMiddleware(predicate=payload_matches("kwargs.env", "prod")),
])

When the middleware predicate denies, only that middleware is skipped for the current job — job dispatch is unaffected.

Gates run once, on the producer side, at enqueue time — there's no worker-dispatch phase. Keep them fast and side-effect-free. Gates are in-process functions: there's no serializable predicate DSL (no JSON/string form, no dashboard round-trip). Configuration can select which registered gate to apply, by name, but not describe a new one.

Predicates and gates run once, synchronously, on the producer thread at enqueue time — there's no worker-dispatch phase. Keep them fast and side-effect-free. Gates are in-process lambdas — there's no serializable predicate DSL (no JSON or string form).

There is no DSL, no JSON or string form and nothing for an operator to edit from a dashboard: a check written here is compiled code, so changing one is a deploy. There is no middleware surface to filter either — the Rust SDK has no hooks.

Where a check runs is therefore entirely yours to choose, and the two places are not equivalent. Before enqueue, a refusal means nothing was written. Inside the task body the job already exists, so its refusal is an outcome: TaskError::fatal to dead-letter it, TaskError::retryable to spend a retry and come back.

Each job in the batch is evaluated independently: a Defer outcome only adjusts that job's own delay and doesn't affect the rest of the batch. Only a Cancel (or False with on_false="cancel") aborts the whole call, since it raises before any job is written.

Inside enqueueMany, only an Allow decision is accepted per entry — a Skip or Defer also aborts the batch with EnqueueSkippedException, since the batch API can't schedule or drop individual entries. Use single enqueue/tryEnqueue calls for payloads that need skip/defer semantics.

Each entry is gated independently and a defer only adjusts that entry's own delay, since every entry carries its own options. A skip aborts the batch with EnqueueSkippedError: the returned job ids line up with the input, so a single dropped entry isn't expressible. Use tryEnqueue per payload when entries need skip semantics.

enqueue_batch evaluates nothing per entry, so the pre-filter is the filter: build the Vec<TaskCall<T>> out of the calls you already decided to make. The refusals it does have are structural rather than policy — an argument that will not encode, a debounce window, a namespace this handle does not own — and each fails the whole call before a single row is written.

Every registered predicate is serialized once at decoration time. The JSON is reachable through:

queue.list_predicates()
# {"app.send_invoice": {"op": "and", "args": [...]}}

queue.predicate_for("app.send_invoice")
# {"op": "and", "args": [...]}

Bare callables (predicate=lambda ctx: …) appear in list_predicates with None as their value, since callables have no stable schema.

The dashboard reads list_predicates() to display "gated by: …" beside each task. To round-trip a predicate from the inspection output:

from flexiq.predicates import Predicate

blob = queue.predicate_for("app.send_invoice")
restored = Predicate.from_dict(blob)
# restored.format() — stable string form
# restored.evaluate(ctx) — re-evaluate against a context

Node gates are closures, not a serializable AST, so there's no listPredicates() equivalent and the dashboard shows no "gated by" schema for Node. The one thing that is introspectable is the registry: defaultRegistry().names() lists every named gate, and has(name) tests one.

Java exposes no predicate inspection API — registered gates are opaque and there is no listPredicates() equivalent.

Nothing is registered, so nothing is persisted and there is no listing to read back. A check written here lives in the source, which is also the only place it can be inspected or changed.

queue.predicate_stats()
# {"allowed": 412, "denied": 3, "deferred": 8, "cancelled": 1, "errors": 0}

Three events fire on the queue's event bus:

• EventType.PREDICATE_DEFERRED — payload includes task_name, defer_seconds, phase ("enqueue" or "dispatch").
• EventType.PREDICATE_CANCELLED — worker-dispatch cancellations.
• EventType.PREDICATE_REJECTED — enqueue-time rejections that raised PredicateRejectedError.

from flexiq.events import EventType

def on_predicate_event(event_type: EventType, payload: dict) -> None:
    print(event_type.value, payload)

for event in (
    EventType.PREDICATE_DEFERRED,
    EventType.PREDICATE_CANCELLED,
    EventType.PREDICATE_REJECTED,
):
    queue.on_event(event, on_predicate_event)

queue.predicateStats();
// { allowed: 412, skipped: 1, deferred: 8, rejected: 3, errors: 0 }

One count per gated enqueue — the decision that won, not one per gate evaluated. Enqueues of ungated tasks aren't counted. The keys follow Node's decision kinds; Python calls rejected denied and skipped cancelled.

Three events fire on the queue's event bus, all carrying taskName plus the gate's reason (rejected / skipped) or delayMs (deferred):

for (const event of ["predicate.rejected", "predicate.skipped", "predicate.deferred"] as const) {
  queue.on(event, (payload) => console.log(event, payload));
}

They're also webhook-subscribable.

Java emits no predicate metrics or events — no predicateStats(), and the event enum carries no deferred/rejected predicate signals.

No counters and no events: with nothing evaluating, there is nothing to count, and the Rust SDK has no event bus to fire on either. What the scheduler's own gates do is visible in the jobs themselves — a job a rate limit deferred stays pending and dispatches later, while one shed by on_excess = "drop" is dead-lettered with a reason naming the limit. queue.stats() counts both.

Business-hours-only report with urgent override

from flexiq.predicates import is_business_hours, payload_matches

@queue.task(
    predicate=is_business_hours(tz="US/Pacific")
              | payload_matches("kwargs.urgent", True),
)
def send_daily_report(team_id: str, urgent: bool = False): ...

send_daily_report.delay("alpha")                        # defers to 09:00 PT
send_daily_report.delay("alpha", urgent=True)           # runs now

Tenant allowlist as a custom predicate

from typing import Any
from flexiq.predicates import Predicate, PredicateContext

@queue.register_predicate("tenant_in")
class TenantIn(Predicate):
    OP = "tenant_in"

    def __init__(self, *, tenants: list[str]) -> None:
        self.tenants = list(tenants)

    def evaluate(self, ctx: PredicateContext) -> bool:
        return ctx.kwargs.get("tenant") in self.tenants

    def to_dict(self) -> dict[str, Any]:
        return {"op": "tenant_in", "tenants": self.tenants}

    @classmethod
    def _from_kwargs(cls, kwargs: dict[str, Any]) -> "Predicate":
        return cls(**kwargs)


@queue.task(
    predicate=TenantIn(tenants=["acme", "globex"]),
    on_false="cancel",
)
def reindex_search(tenant: str): ...

Async predicate calling an HTTP quota service

import httpx
from flexiq.predicates import Defer, Predicate, PredicateContext

class QuotaAvailable(Predicate):
    OP = "quota_available"

    def __init__(self, *, url: str) -> None:
        self.url = url

    async def evaluate(self, ctx: PredicateContext) -> bool | Defer:
        async with httpx.AsyncClient(timeout=2.0) as c:
            r = await c.get(f"{self.url}/quota/{ctx.kwargs['tenant']}")
        if r.json()["remaining"] > 0:
            return True
        return Defer(seconds=60.0)


queue.register_predicate("quota_available")(QuotaAvailable)

@queue.task(predicate=QuotaAvailable(url="http://quota.internal"))
async def index_document(tenant: str, doc_id: str): ...

Tenant allowlist (deny rejects the enqueue)

const allowed = new Set(["acme", "globex"]);

queue.gate("reindexSearch", ({ args }) => allowed.has(args[0] as string));

queue.enqueue("reindexSearch", ["acme"]); // ok
queue.enqueue("reindexSearch", ["evilcorp"]); // throws PredicateRejectedError

Business-hours + urgent override (recipe defers, urgent allows now)

import { anyOf, businessHours } from "@byteveda/flexiq";

const urgent = ({ args }: { args: readonly unknown[] }) => args[1] === true;

queue.gate("sendDailyReport", anyOf(urgent, businessHours({ timeZone: "America/Los_Angeles" })));

queue.enqueue("sendDailyReport", ["alpha"]); // out of hours: deferred to the next 09:00 PT
queue.enqueue("sendDailyReport", ["alpha", true]); // urgent bypass: runs now

Quietly dropping work a flag has turned off

import { Recipes } from "@byteveda/flexiq";

queue.gate("exportLedger", Recipes.featureFlag("beta-export"));

queue.tryEnqueue("exportLedger", ["acme"]); // flag off: null, no job created
queue.enqueue("exportLedger", ["acme"]); // flag off: throws EnqueueSkippedError

Business-hours + urgent override (recipe defers, urgent allows now)

EnqueueGate hours = Recipes.businessHours(ZoneId.of("America/Los_Angeles"));

flexiq.gate("sendDailyReport", ctx ->
    ((Report) ctx.payload()).urgent() ? EnqueueDecision.allow() : hours.decide(ctx));

flexiq.enqueue(sendReport, new Report("alpha", false)); // defers to next 09:00 PT
flexiq.enqueue(sendReport, new Report("alpha", true));  // runs now

Tenant allowlist (skip vs reject, gate-aware tryEnqueue)

Set<String> allowed = Set.of("acme", "globex");

flexiq.gate("reindexSearch", ctx ->
    allowed.contains(((Reindex) ctx.payload()).tenant())
        ? EnqueueDecision.allow()
        : EnqueueDecision.reject("tenant not allowed"));

Optional<String> id = flexiq.tryEnqueue(reindex, new Reindex("acme")); // present
flexiq.enqueue(reindex, new Reindex("evilcorp"));   // throws PredicateRejectedException

Each of these is the same rule the other tiers register as a gate, written where the value already is.

Business hours, with an urgent override

A defer is a delay, so the rule resolves to one number:

#[flexiq::task(queue = "reports")]
fn send_daily_report(team_id: String, urgent: bool) -> flexiq::Outcome<()> {
    Ok(())
}

// `ms_until_open` is yours: 0 inside the window, otherwise the wait until 09:00.
let wait = if urgent { 0 } else { ms_until_open("America/Los_Angeles") };
queue.enqueue(send_daily_report::call(team_id, urgent).delay_ms(wait))?;

Tenant allowlist, refused before anything is written

fn reindex(queue: &FlexiQ, tenant: String) -> flexiq::Result<Option<flexiq::Job>> {
    if !allowed.contains(tenant.as_str()) {
        // No row, no claim, nothing to cancel.
        return Ok(None);
    }
    Ok(Some(queue.enqueue(reindex_search::call(tenant))?))
}

A feature flag read at dispatch instead

Sometimes the flag has to be read when the job runs, not when it is created. A refusal there is an outcome, and it is fatal rather than retryable because another attempt will not turn the flag on:

#[flexiq::task]
fn export_ledger(tenant: String) -> flexiq::Outcome<()> {
    if !flag_enabled("beta-export") {
        return Err(flexiq::TaskError::fatal("beta-export is disabled").into());
    }
    Ok(())
}