Predicates

Composable gates that decide whether a job runs — from a simple boolean check to a serializable rule.

A predicate (also called a gate) is a small check evaluated before a job is allowed to proceed — a declarative way to say "only do this under condition X." What a denial does depends on the configured outcome: a reject or cancel outcome means no work runs, while a defer outcome schedules the job to try again later.

Register one with the predicate= option on @queue.task.Register one with queue.gate(), returning a boolean or a richer Decision.Register one with flexiq.predicate(), or the richer flexiq.gate().

Quick start

python

from flexiq import Queue

queue = Queue()

@queue.task(predicate=lambda ctx: ctx.args[0] > 0, on_false="cancel")
def charge(amount: float):
    settle(amount)

charge.delay(50)   # ok
charge.delay(-1)   # raises PredicateRejectedError

queue.task("charge", (amount: number) => settle(amount));

queue.gate("charge", ({ args }) => args[0] > 0); // never enqueue a non-positive charge

queue.enqueue("charge", [50]); // ok
queue.enqueue("charge", [-1]); // throws PredicateRejectedError

java

Task<Integer> charge = Task.of("charge", Integer.class);

flexiq.predicate("charge", ctx -> (Integer) ctx.payload() > 0);

flexiq.enqueue(charge, 50);  // ok
flexiq.enqueue(charge, -1);  // throws PredicateRejectedException

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:

Goal	Use this — not a predicate
Cap concurrent executions of a task	`@queue.task(max_concurrent=N)`
Cap concurrent executions of a queue	`Queue.set_queue_concurrency(name, N)`
Rate-limit a task or queue	`@queue.task(rate_limit="100/m")`
Trip an outage breaker	`@queue.task(circuit_breaker={...})`
Drain / pause a queue	`Queue.pause(name)` / `resume(name)`
Cap retry count or filter retries	`max_retries=`, `retry_on=`, `dont_retry_on=`
Route to a queue	`@queue.task(queue="emails")`

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.

Composition

python

allow = is_business_hours() & ~queue_paused()
allow_or_urgent = allow | feature_flag("urgent_bypass")

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

const highPriority = ({ args }) => args[0].priority === "high";

queue.gate("dispatch", anyOf(businessHours({ timeZone: "UTC" }), highPriority));
queue.gate("dispatch", not(isWeekend()));

allOf / anyOf accept decision-returning gates too: a blocking decision survives the composition, so the anyOf above still defers out-of-hours work rather than rejecting it (not is boolean-only — inverting a defer has no meaning).

java

Predicate positive = ctx -> (Integer) ctx.payload() > 0;
Predicate small = ctx -> (Integer) ctx.payload() <= 10_000;

flexiq.predicate("charge", Predicates.allOf(positive, small));
flexiq.predicate("dispatch", Predicates.not(isHoliday));

@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.

Richer outcomes: defer, cancel, skip

A plain boolean allow/deny is the baseline every binding supports.

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

Outcome	Meaning
`True`	Allow the job to proceed.
`False`	Deny. The task's `on_false` decides: `"defer"` (default) or `"cancel"`.
`Defer(seconds=N)`	Skip now, retry after `N` seconds.
`Cancel(reason="…")`	Permanently skip. Raises `PredicateRejectedError` at enqueue; cancels the job at dispatch.

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.

python

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:

Outcome	Meaning
`true` / `Decision.allow()`	Enqueue unchanged.
`false` / `Decision.reject(reason)`	Block it: `enqueue` throws `PredicateRejectedError`, and so does `tryEnqueue`.
`Decision.skip(reason)`	Quietly don't enqueue: `enqueue` throws `EnqueueSkippedError`, `tryEnqueue` returns `null`.
`Decision.defer(ms)`	Enqueue delayed by `ms`, replacing any `delayMs` the caller passed.
`Decision.deferUntil(date)`	The same, expressed as an instant (a past instant means no delay).

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.

java

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");
});

Decision	`enqueue`	`tryEnqueue`
`allow()`	Job created.	Job id present.
`skip(reason)`	Throws `EnqueueSkippedException`.	Returns empty.
`defer(delay)`	Job created, scheduled `delay` from now (overrides any delay in the options).	Job id present.
`deferUntil(instant)`	Job created, scheduled at the absolute `instant` (overrides any delay in the options).	Job id present.
`reject(reason)`	Throws `PredicateRejectedException`.	Throws too.

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

java

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.

Built-in recipes

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

Time

Recipe	Description
`is_business_hours(start=9, end=17, *, tz=None, weekdays_only=True)`	Allow during business hours; defer to the next opening otherwise.
`is_weekend(*, tz=None)`	True on Sat/Sun.
`in_time_window("09:00", "17:00", *, tz=None)`	Allow within `[start, end)`; defer otherwise.
`after(target)`	Defer until `target` (`datetime` or ISO string).
`before(target)`	Allow only strictly before `target`.
`in_timezone(tz)`	No-op at runtime; validates the tz string at construction time.

Payload

Recipe	Description
`payload_matches("kwargs.tenant", "acme")`	Dotted-path lookup into `{"args": …, "kwargs": …}`; compare against `expected`.

Defensive

Recipe	Description
`queue_paused(name=None)`	Read pause state. Use inverted (`~queue_paused()`) as a defensive composable, not as primary enforcement (which is `Queue.pause()` / `resume()`).

External config

Recipe	Description
`env_var_truthy("MY_FLAG")`	Truthy values: `1`, `true`, `t`, `yes`, `y`, `on` (case-insensitive).
`feature_flag("billing", *, provider=None)`	Defaults to env-var provider reading `FF_<NAME>`. Pass a registered provider name to swap in LaunchDarkly / Statsig / etc.

Pluggable feature-flag provider

python

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:

java

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.

Recipe	Description
`businessHours({ timeZone?, startHour = 9, endHour = 17, weekdaysOnly = true })`	Allow inside the window; defer to the next opening otherwise.
`timeWindow({ start: "22:00", end: "06:00", timeZone? })`	Allow within `[start, end)`; defer to the next `start`. An `end` before `start` wraps midnight.
`dayOfWeek({ days: ["mon", "thu"], timeZone? })`	Allow on those days; defer to the start of the next allowed day. No days means every enqueue skips.
`isWeekend({ timeZone? })`	Boolean: true on Sat/Sun. Usually used as `not(isWeekend())`.
`after(target)`	Defer until `target` (a `Date` or ISO string).
`before(target)`	Boolean: true only strictly before `target`.

Payload

Recipe	Description
`payloadMatches("args.0.tenantId", "acme")`	Dotted-path lookup rooted at `{ args }`, compared with `Object.is`. Skips on a mismatch or a missing path.

External config

Recipe	Description
`envVarTruthy("MY_FLAG")`	Boolean. Truthy values: `1`, `true`, `t`, `yes`, `y`, `on` (case-insensitive).
`featureFlag("beta-export", provider?)`	Skips while the flag is off. Defaults to `envFeatureFlagProvider()`, reading `FF_BETA_EXPORT`.

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.

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:

python

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"},
    ],
})

python

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:

python

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:

python

@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:

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.
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.

python

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).

Batches

enqueue_many evaluates the task's predicateenqueueMany gates each entryenqueueMany gates each payload — a single rejection aborts the whole batch and no jobs are created. Pre-filter the batch yourself if you'd rather drop rejected entries silently.

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.

Gates run on the producer side, so they decide whether work is created at all. To react once a job is already running, use middleware or conditions inside a workflow.

Persistence & inspection

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

python

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:

python

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.

Metrics & events

python

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.

python

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.

Examples

Business-hours-only report with urgent override

python

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

python

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

python

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)

java

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`)

java

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