Wire message deferral tokens through AMQP 1.0 FLOW to quorum queues

Consumers can now request previously deferred (parked) messages by
listing their tokens under the rabbitmq:deferred-tokens FLOW property.
rabbit_queue_type gains an optional assign_deferred/4 callback,
implemented by quorum queues via rabbit_fifo_client. In rabbit_fifo,
a deferral token is only honoured when the client also sets an
explicit x-opt-delivery-time; the delayed-retry path never creates a
deferred entry, keeping tokens purely client-driven.

Advertise support via the rabbitmq:deferred-tokens link capability
and document the protocol in deps/rabbit/DEFERRED.md.

Author: kjnilsson

deps/rabbit/DEFERRED.md

@@ -0,0 +1,112 @@
+# Message Deferral for Quorum Queues
+
+## Overview
+
+Message deferral lets an AMQP 1.0 consumer return a message to a quorum queue in a
+*parked* state under a client-chosen token, then later pull that specific message back
+on demand — without waiting for a delayed-retry timer or competing with other consumers.
+
+This is useful for workload scheduling patterns where a consumer wants to decide at
+receive time that a message should not be processed immediately, assign it a token
+for later retrieval, and then fetch it explicitly when ready.
+
+## Constraints
+
+- **Quorum queues only.** Classic queues and streams do not support deferral tokens.
+  Clients can detect support by checking for `rabbitmq:deferral-tokens` in the
+  `offered-capabilities` field of the ATTACH response.
+- **`x-opt-delivery-time` is required.** A deferral token alone does not park a
+  message. Both `x-opt-deferral-token` and `x-opt-delivery-time` must be present in
+  the MODIFIED outcome's `message-annotations`. A message returned via the
+  `delayed-retry` queue configuration never creates a deferred entry, even if a token
+  is present.
+- **Tokens must be of AMQP type `utf8`.** This applies both to `x-opt-deferral-token`
+  in the MODIFIED outcome's `message-annotations` and to each element of the
+  `rabbitmq:deferral-tokens` array in a FLOW frame's `properties`. Any other AMQP
+  type (e.g. `binary`, `symbol`) is rejected with `amqp:invalid-field`.
+- **Credit must cover the matched messages.** When requesting deferred messages via a
+  FLOW frame, the link credit granted in that same FLOW must be at least as large as
+  the number of messages the submitted tokens resolve to (a single token may resolve
+  to more than one message; see below). If credit is exhausted by normal message
+  delivery before the deferred assignment runs, no deferred messages are delivered
+  for that FLOW.
+
+## Protocol Usage
+
+### 1. Attach a consuming link and verify capability
+
+Attach a link to a quorum queue source. Inspect the `offered-capabilities` array in
+the ATTACH response for the symbol `rabbitmq:deferral-tokens`. If absent, the queue
+does not support deferral.
+
+### 2. Receive a message
+
+The broker delivers a message to the link via a TRANSFER frame. Note the
+`delivery-tag` for the settlement step.
+
+### 3. Park the message with a deferral token
+
+Send a DISPOSITION frame settling the delivery with a MODIFIED outcome. Include both
+annotations in the `message-annotations` map of the MODIFIED outcome:
+
+- `x-opt-deferral-token` (symbol key, utf8 value) — a client-chosen opaque
+  identifier for this parked message. The same token may be assigned to more than
+  one message, e.g. by settling a range of deliveries (`first =/= last`) with a
+  single MODIFIED outcome. Retrieving such a token returns all messages parked
+  under it, oldest first.
+- `x-opt-delivery-time` (symbol key, timestamp value in milliseconds since the Unix
+  epoch) — the earliest time at which the message becomes eligible for normal
+  timer-based redelivery. The message is held until this time unless explicitly
+  retrieved earlier via `assign_deferred`.
+
+Example (pseudocode):
+
+```
+DISPOSITION {
+  role = receiver,
+  first = <delivery-id>,
+  settled = true,
+  state = MODIFIED {
+    delivery-failed = false,
+    undeliverable-here = false,
+    message-annotations = {
+      x-opt-deferral-token: "job-42-retry-1",
+      x-opt-delivery-time: 1780000000000
+    }
+  }
+}
+```
+
+### 4. Retrieve the message by token
+
+When ready to process the parked message, send a FLOW frame on the same consuming
+link. Grant enough link credit to receive the deferred messages and include the tokens
+in the `properties` map of the FLOW frame under the key `rabbitmq:deferral-tokens` as
+an array of utf8 values.
+
+Example (pseudocode):
+
+```
+FLOW {
+  handle = <link-handle>,
+  delivery-count = <current-delivery-count>,
+  link-credit = 1,
+  properties = {
+    rabbitmq:deferral-tokens: ["job-42-retry-1"]
+  }
+}
+```
+
+The broker looks up each token in the parked message set and delivers any matches
+directly to this link as normal TRANSFER frames. Tokens that are not found (because
+the message was already expired by its delivery time and requeued normally, or the
+token was never issued) produce no delivery — the client is responsible for tracking
+which tokens it expects.
+
+## Relationship to `delayed-retry`
+
+Deferral tokens and the queue-level `x-delayed-retry-*` configuration are
+complementary but independent. A message returned with `x-opt-delivery-time` follows
+the deferral path. A message returned without `x-opt-delivery-time` follows the
+delayed-retry path if the queue is configured for it. The two paths do not interact:
+delayed-retry messages cannot be retrieved by token.

deps/rabbit/src/rabbit_amqp_session.erl

@@ -2148,7 +2148,14 @@ settle_op_from_outcome(#'v1_0.modified'{delivery_failed = DelFailed,
                {map, KVList} ->
                    Anns1 = lists:map(
                              %% "all symbolic keys except those beginning with "x-" are reserved." [3.2.10]
-                             fun({{symbol, <<"x-", _/binary>> = K}, V}) ->
+                             fun({{symbol, <<"x-opt-deferral-token">> = K}, {utf8, _} = V}) ->
+                                     {K, unwrap_simple_type(V)};
+                                ({{symbol, <<"x-opt-deferral-token">>}, Other}) ->
+                                     protocol_error(
+                                       ?V_1_0_AMQP_ERROR_INVALID_FIELD,
+                                       "x-opt-deferral-token must be of AMQP type utf8, got: ~tp",
+                                       [Other]);
+                                ({{symbol, <<"x-", _/binary>> = K}, V}) ->
                                      {K, unwrap_simple_type(V)}
                              end, KVList),
                    maps:from_list(Anns1)
@@ -3162,7 +3169,8 @@ handle_outgoing_link_flow_control(
                delivery_count = MaybeDeliveryCountRcv,
                link_credit = ?UINT(LinkCreditRcv),
                drain = Drain0,
-               echo = Echo0},
+               echo = Echo0,
+               properties = FlowProps},
   #state{outgoing_links = OutgoingLinks,
          queue_states = QStates0
         } = State0) ->
@@ -3186,14 +3194,21 @@ handle_outgoing_link_flow_control(
                                         credit = CappedCredit,
                                         drain = Drain},
                      at_least_one_credit_req_in_flight = true},
-            {ok, QStates, Actions} = rabbit_queue_type:credit(
-                                       QName, Ctag,
-                                       QFC#queue_flow_ctl.delivery_count,
-                                       CappedCredit, Drain, QStates0),
+            {ok, QStates1, Actions0} = rabbit_queue_type:credit(
+                                         QName, Ctag,
+                                         QFC#queue_flow_ctl.delivery_count,
+                                         CappedCredit, Drain, QStates0),
+            {ok, QStates, Actions1} = case parse_deferred_tokens(FlowProps) of
+                                          [] ->
+                                              {ok, QStates1, []};
+                                          Tokens ->
+                                              rabbit_queue_type:assign_deferred(
+                                                QName, Ctag, Tokens, QStates1)
+                                      end,
             State = State0#state{
                       queue_states = QStates,
                       outgoing_links = OutgoingLinks#{HandleInt := Link}},
-            handle_queue_actions(Actions, State);
+            handle_queue_actions(Actions0 ++ Actions1, State);
         true ->
             %% A credit request is currently in-flight. Let's first process its reply
             %% before sending the next request. This ensures our outgoing_pending
@@ -3203,6 +3218,8 @@ handle_outgoing_link_flow_control(
             %% to reason about. Therefore, we stash the new request. If there is already a
             %% stashed request, we replace it because the latest flow control state from the
             %% client applies.
+            %% Deferral tokens are not stashed; they target the current state of the delayed
+            %% set and must not be replayed when the stashed credit request is processed.
             Link = Link0#outgoing_link{
                      stashed_credit_req = #credit_req{
                                              delivery_count = DeliveryCountRcv,
@@ -3212,6 +3229,21 @@ handle_outgoing_link_flow_control(
             State0#state{outgoing_links = OutgoingLinks#{HandleInt := Link}}
     end.
 
+parse_deferred_tokens(undefined) ->
+    [];
+parse_deferred_tokens({map, KVList}) ->
+    case lists:keyfind({symbol, <<"rabbitmq:deferral-tokens">>}, 1, KVList) of
+        {{symbol, <<"rabbitmq:deferral-tokens">>}, {array, utf8, Elems}} ->
+            [T || {utf8, T} <- Elems];
+        false ->
+            [];
+        {{symbol, <<"rabbitmq:deferral-tokens">>}, Other} ->
+            protocol_error(
+              ?V_1_0_AMQP_ERROR_INVALID_FIELD,
+              "rabbitmq:deferral-tokens must be an array of AMQP type utf8, got: ~tp",
+              [Other])
+    end.
+
 delivery_count_rcv(?UINT(DeliveryCount)) ->
     DeliveryCount;
 delivery_count_rcv(undefined) ->

deps/rabbit/src/rabbit_fifo.erl

@@ -2403,24 +2403,20 @@ return_one(#{system_time := Ts} = Meta, MsgId,
     end.
 
 should_delay(DeliveryFailed, DelayedRetry, Ts, Header, Anns) ->
-    %% First check for explicit x-opt-delivery-time annotation.
-    %% This takes precedence over delayed_retry configuration.
-    DeferralToken = case Anns of
-                        #{<<"x-opt-deferral-token">> := Token}
-                          when is_binary(Token) ->
-                            Token;
-                        _ ->
-                            undefined
-                    end,
     case Anns of
         #{<<"x-opt-delivery-time">> := DeliveryTime}
           when is_integer(DeliveryTime),
                DeliveryTime > Ts ->
+            %% Deferral tokens are only honoured when the client explicitly
+            %% sets a delivery time; the delayed-retry path never creates a
+            %% deferred entry so that tokens remain a purely client-driven
+            %% mechanism.
+            DeferralToken = maps:get(<<"x-opt-deferral-token">>, Anns, undefined),
             {true, DeliveryTime, DeferralToken};
         _ ->
             case should_delay0(DeliveryFailed, DelayedRetry, Ts, Header) of
                 {true, ReadyAt} ->
-                    {true, ReadyAt, DeferralToken};
+                    {true, ReadyAt, undefined};
                 false ->
                     false
             end
@@ -2616,8 +2612,7 @@ take_next_delayed(Ts, #delayed{next = {ReadyAt, Idx, Msg},
                    {?TUPLE(NextReadyAt, NextIdx), V} = gb_trees:smallest(Tree),
                    {NextReadyAt, NextIdx, V}
            end,
-    %% Remove any deferral token that maps to this key
-    Deferred = maps:filter(fun(_Token, K) -> K =/= Key end, Deferred0),
+    Deferred = remove_deferred_key(Key, Deferred0),
     Delayed = #delayed{tree = Tree, next = Next, deferred = Deferred},
     {Msg, Delayed};
 take_next_delayed(_Ts, #delayed{}) ->
@@ -2660,12 +2655,22 @@ take_delayed_for_retry(N, Ts, #delayed{tree = Tree0,
                            ?TUPLE(ReadyAt, Idx) = NextKey,
                            {ReadyAt, Idx, NextMsg}
                    end,
-            %% Remove any deferral token that maps to this key
-            Deferred = maps:filter(fun(_Token, K) -> K =/= Key end, Deferred0),
+            Deferred = remove_deferred_key(Key, Deferred0),
             Delayed = #delayed{tree = Tree, next = Next, deferred = Deferred},
             take_delayed_for_retry(N - 1, Ts, Delayed, [Msg | Acc])
     end.
 
+%% Drop a single tree key from every token's key list, dropping the token
+%% entirely once its last key is removed.
+remove_deferred_key(Key, Deferred0) ->
+    maps:filtermap(
+      fun(_Token, Keys) ->
+              case lists:delete(Key, Keys) of
+                  [] -> false;
+                  Remaining -> {true, Remaining}
+              end
+      end, Deferred0).
+
 take_deferred(Tokens, Delayed) ->
     take_deferred(Tokens, Delayed, [], []).
 
@@ -2675,20 +2680,30 @@ take_deferred([Token | Rest], #delayed{tree = Tree0,
                                        deferred = Deferred0} = Delayed0,
               MsgsAcc, NotFoundAcc) ->
     case maps:take(Token, Deferred0) of
-        {Key, Deferred1} ->
-            case gb_trees:lookup(Key, Tree0) of
-                {value, Msg} ->
-                    Tree = gb_trees:delete(Key, Tree0),
-                    Next = update_delayed_next(Tree),
-                    Delayed = Delayed0#delayed{tree = Tree,
-                                               next = Next,
-                                               deferred = Deferred1},
-                    take_deferred(Rest, Delayed, [Msg | MsgsAcc], NotFoundAcc);
-                none ->
-                    %% Key in deferred map but not in tree - inconsistent,
-                    %% treat as not found and clean up
-                    Delayed = Delayed0#delayed{deferred = Deferred1},
-                    take_deferred(Rest, Delayed, MsgsAcc, [Token | NotFoundAcc])
+        {Keys, Deferred1} ->
+            %% Keys were prepended as they were parked, so reverse to
+            %% resolve them oldest first.
+            {Tree, MsgsAcc1, Found} =
+                lists:foldl(
+                  fun(Key, {TreeAcc, Acc, FoundAcc}) ->
+                          case gb_trees:lookup(Key, TreeAcc) of
+                              {value, Msg} ->
+                                  {gb_trees:delete(Key, TreeAcc), [Msg | Acc], true};
+                              none ->
+                                  %% Key in deferred map but not in tree -
+                                  %% inconsistent, skip and clean up
+                                  {TreeAcc, Acc, FoundAcc}
+                          end
+                  end, {Tree0, MsgsAcc, false}, lists:reverse(Keys)),
+            Next = update_delayed_next(Tree),
+            Delayed = Delayed0#delayed{tree = Tree,
+                                       next = Next,
+                                       deferred = Deferred1},
+            case Found of
+                true ->
+                    take_deferred(Rest, Delayed, MsgsAcc1, NotFoundAcc);
+                false ->
+                    take_deferred(Rest, Delayed, MsgsAcc1, [Token | NotFoundAcc])
             end;
         error ->
             take_deferred(Rest, Delayed0, MsgsAcc, [Token | NotFoundAcc])
@@ -2765,7 +2780,9 @@ delayed_in(ReadyAt, Idx, Msg, DeferralToken, #delayed{tree = Tree0,
                    undefined ->
                        Deferred0;
                    _ ->
-                       Deferred0#{DeferralToken => Key}
+                       maps:update_with(DeferralToken,
+                                        fun(Keys) -> [Key | Keys] end,
+                                        [Key], Deferred0)
                end,
     #delayed{tree = Tree, next = Next, deferred = Deferred}.
 

deps/rabbit/src/rabbit_fifo.hrl

@@ -202,8 +202,11 @@
         {tree = gb_trees:empty() :: gb_trees:tree(delayed_key(), msg()),
          %% Cached smallest entry for O(1) readiness check in take_next_msg
          next = undefined :: option({milliseconds(), ra:index(), msg()}),
-         %% Map from deferral token to tree key for direct message lookup
-         deferred = #{} :: #{deferral_token() => delayed_key()}}).
+         %% Map from deferral token to the tree keys of all messages parked
+         %% under it. A single token may address more than one message,
+         %% e.g. when a ranged DISPOSITION settles several deliveries with
+         %% the same annotations.
+         deferred = #{} :: #{deferral_token() => [delayed_key()]}}).
 
 -record(enqueuer,
         {next_seqno = 1 :: msg_seqno(),

deps/rabbit/src/rabbit_fifo_client.erl

@@ -527,26 +527,16 @@ retry_delayed(Server, Mode) when Mode =:= all orelse
 %% @param ConsumerTag the tag uniquely identifying the consumer.
 %% @param Tokens list of deferral tokens (from x-opt-deferral-token).
 %% @param State the {@module} state
-%% @returns `{ok, NumAssigned}' on success,
-%%   `{partial, NumAssigned, NotFoundTokens}' if some tokens not found,
-%%   `{error, Reason}' on failure.
+%% @returns `{state(), actions()}' - matched messages arrive as normal deliveries.
 -spec assign_deferred(rabbit_types:ctag(), [binary()], state()) ->
-    {{ok, non_neg_integer()} |
-     {partial, non_neg_integer(), [binary()]} |
-     {error, term()},
-     state()}.
+    {state(), rabbit_queue_type:actions()}.
 assign_deferred(ConsumerTag, [_|_] = Tokens, State0) ->
     ConsumerKey = consumer_key(ConsumerTag, State0),
     ServerId = pick_server(State0),
     Cmd = rabbit_fifo:make_delayed({assign_deferred, ConsumerKey, Tokens}),
-    case ra:process_command(ServerId, Cmd, ?COMMAND_TIMEOUT) of
-        {ok, Reply, _} ->
-            {Reply, State0};
-        Err ->
-            {Err, State0}
-    end;
+    {send_command(ServerId, undefined, Cmd, normal, State0), []};
 assign_deferred(_ConsumerTag, [], State) ->
-    {{ok, 0}, State}.
+    {State, []}.
 
 -spec pending_size(state()) -> non_neg_integer().
 pending_size(#state{pending = Pend}) ->