merge with default

11 files changed, 578 insertions, 258 deletions

diff --git a/include/rabbit.hrl b/include/rabbit.hrl
index 24aa8d98..0f0a0e87 100644
--- a/include/rabbit.hrl
+++ b/include/rabbit.hrl
@@ -74,6 +74,8 @@
 
 -record(event, {type, props, timestamp}).
 
+-record(msg_properties, {expiry}).
+
 %%----------------------------------------------------------------------------
 
 -define(COPYRIGHT_MESSAGE, "Copyright (C) 2007-2010 LShift Ltd., Cohesive Financial Technologies LLC., and Rabbit Technologies Ltd.").
diff --git a/include/rabbit_backing_queue_spec.hrl b/include/rabbit_backing_queue_spec.hrl
index 005994f0..f750fbb2 100644
--- a/include/rabbit_backing_queue_spec.hrl
+++ b/include/rabbit_backing_queue_spec.hrl
@@ -30,12 +30,18 @@
 %%
 
 -type(fetch_result() ::
-                 %% Message,  IsDelivered,  AckTag,  Remaining_Len
-        ('empty'|{rabbit_types:basic_message(), boolean(), ack(), non_neg_integer()})).
+                 %% Message, IsDelivered,  AckTag,  Remaining_Len
+        ('empty'|{rabbit_types:basic_message(), 
+                  boolean(), ack(), 
+                  non_neg_integer()})).
+-type(peek_result() :: ('empty'|{rabbit_types:basic_message(), 
+                                 rabbit_types:msg_properties()})).
 -type(is_durable() :: boolean()).
 -type(attempt_recovery() :: boolean()).
 -type(purged_msg_count() :: non_neg_integer()).
 -type(ack_required() :: boolean()).
+-type(msg_properties_transformer() :: 
+        fun ((rabbit_types:msg_properties()) -> rabbit_types:msg_properties())).
 
 -spec(start/1 :: ([rabbit_amqqueue:name()]) -> 'ok').
 -spec(stop/0 :: () -> 'ok').
@@ -43,16 +49,26 @@
 -spec(terminate/1 :: (state()) -> state()).
 -spec(delete_and_terminate/1 :: (state()) -> state()).
 -spec(purge/1 :: (state()) -> {purged_msg_count(), state()}).
--spec(publish/2 :: (rabbit_types:basic_message(), state()) -> state()).
--spec(publish_delivered/3 ::
-        (ack_required(), rabbit_types:basic_message(), state()) -> {ack(), state()}).
+-spec(publish/3 :: 
+        (rabbit_types:basic_message(), rabbit_types:msg_properties(), state()) 
+        -> state()).
+-spec(publish_delivered/4 ::
+        (ack_required(), rabbit_types:basic_message(), 
+         rabbit_types:msg_properties(), state()) -> {ack(), state()}).
+-spec(dropwhile/2 :: 
+        (fun ((rabbit_types:msg_properties()) -> boolean()), state()) 
+        -> state()).
 -spec(fetch/2 :: (ack_required(), state()) -> {fetch_result(), state()}).
 -spec(ack/2 :: ([ack()], state()) -> state()).
--spec(tx_publish/3 :: (rabbit_types:txn(), rabbit_types:basic_message(), state()) -> state()).
+-spec(tx_publish/4 :: 
+        (rabbit_types:txn(), rabbit_types:basic_message(), 
+         rabbit_types:msg_properties(), state()) -> state()).
 -spec(tx_ack/3 :: (rabbit_types:txn(), [ack()], state()) -> state()).
 -spec(tx_rollback/2 :: (rabbit_types:txn(), state()) -> {[ack()], state()}).
--spec(tx_commit/3 :: (rabbit_types:txn(), fun (() -> any()), state()) -> {[ack()], state()}).
--spec(requeue/2 :: ([ack()], state()) -> state()).
+-spec(tx_commit/4 :: 
+        (rabbit_types:txn(), fun (() -> any()), 
+         msg_properties_transformer(), state()) -> {[ack()], state()}).
+-spec(requeue/3 :: ([ack()], msg_properties_transformer(), state()) -> state()).
 -spec(len/1 :: (state()) -> non_neg_integer()).
 -spec(is_empty/1 :: (state()) -> boolean()).
 -spec(set_ram_duration_target/2 ::
diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl
index 42bddc5e..83a13f2c 100644
--- a/src/rabbit_amqqueue.erl
+++ b/src/rabbit_amqqueue.erl
@@ -56,7 +56,7 @@
 -include("rabbit.hrl").
 -include_lib("stdlib/include/qlc.hrl").
 
--define(EXPIRES_TYPES, [byte, short, signedint, long]).
+-define(INTEGER_ARG_TYPES, [byte, short, signedint, long]).
 
 %%----------------------------------------------------------------------------
 
@@ -312,19 +312,31 @@ check_declare_arguments(QueueName, Args) ->
                              precondition_failed,
                              "invalid arg '~s' for ~s: ~w",
                              [Key, rabbit_misc:rs(QueueName), Error])
-     end || {Key, Fun} <- [{<<"x-expires">>, fun check_expires_argument/1}]],
+     end || {Key, Fun} <- 
+                [{<<"x-expires">>, fun check_expires_argument/1},
+                 {<<"x-message-ttl">>, fun check_message_ttl_argument/1}]],
     ok.
 
-check_expires_argument(undefined) ->
+check_expires_argument(Val) ->
+    check_integer_argument(Val, 
+                           expires_not_of_acceptable_type, 
+                           expires_zero_or_less).
+
+check_message_ttl_argument(Val) ->
+    check_integer_argument(Val, 
+                           ttl_not_of_acceptable_type, 
+                           ttl_zero_or_less).
+
+check_integer_argument(undefined, _, _) ->
     ok;
-check_expires_argument({Type, Expires}) when Expires > 0 ->
-    case lists:member(Type, ?EXPIRES_TYPES) of
+check_integer_argument({Type, Val}, InvalidTypeError, _) when Val > 0 ->
+    case lists:member(Type, ?INTEGER_ARG_TYPES) of
         true  -> ok;
-        false -> {error, {expires_not_of_acceptable_type, Type, Expires}}
+        false -> {error, {InvalidTypeError, Type, Val}}
     end;
-check_expires_argument({_Type, _Expires}) ->
-    {error, expires_zero_or_less}.
-
+check_integer_argument({_Type, _Val}, _, ZeroOrLessError) ->
+    {error, ZeroOrLessError}.
+  
 list(VHostPath) ->
     mnesia:dirty_match_object(
       rabbit_queue,
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl
index 2c53a8e3..b2519b7a 100644
--- a/src/rabbit_amqqueue_process.erl
+++ b/src/rabbit_amqqueue_process.erl
@@ -39,6 +39,8 @@
 -define(SYNC_INTERVAL,                 5). %% milliseconds
 -define(RAM_DURATION_UPDATE_INTERVAL,  5000).
 
+-define(BASE_MSG_PROPERTIES, #msg_properties{expiry = undefined}).
+
 -export([start_link/1, info_keys/0]).
 
 -export([init/1, terminate/2, code_change/3, handle_call/3, handle_cast/2,
@@ -61,7 +63,9 @@
             sync_timer_ref,
             rate_timer_ref,
             expiry_timer_ref,
-            stats_timer
+            stats_timer,
+            ttl,
+            ttl_timer_ref
            }).
 
 -record(consumer, {tag, ack_required}).
@@ -123,6 +127,7 @@ init(Q) ->
             sync_timer_ref      = undefined,
             rate_timer_ref      = undefined,
             expiry_timer_ref    = undefined,
+            ttl                 = undefined,
             stats_timer         = rabbit_event:init_stats_timer()}, hibernate,
      {backoff, ?HIBERNATE_AFTER_MIN, ?HIBERNATE_AFTER_MIN, ?DESIRED_HIBERNATE}}.
 
@@ -145,12 +150,22 @@ code_change(_OldVsn, State, _Extra) ->
 
 %%----------------------------------------------------------------------------
 
+init_queue_state(State) ->
+    lists:foldl(fun(F, S) -> F(S) end, State, 
+                [fun init_expires/1, fun init_ttl/1]).
+
 init_expires(State = #q{q = #amqqueue{arguments = Arguments}}) ->
     case rabbit_misc:table_lookup(Arguments, <<"x-expires">>) of
         {_Type, Expires} -> ensure_expiry_timer(State#q{expires = Expires});
         undefined        -> State
     end.
 
+init_ttl(State = #q{q = #amqqueue{arguments = Arguments}}) ->
+    case rabbit_misc:table_lookup(Arguments, <<"x-message-ttl">>) of
+        {_Type, Ttl} -> State#q{ttl = Ttl};
+        undefined    -> State
+    end.
+
 declare(Recover, From,
         State = #q{q = Q = #amqqueue{name = QName, durable = IsDurable},
                    backing_queue = BQ, backing_queue_state = undefined,
@@ -165,7 +180,8 @@ declare(Recover, From,
                             self(), {rabbit_amqqueue,
                                      set_ram_duration_target, [self()]}),
                      BQS = BQ:init(QName, IsDurable, Recover),
-                     State1 = init_expires(State#q{backing_queue_state = BQS}),
+                     State1 = init_queue_state(
+                                State#q{backing_queue_state = BQS}),
                      rabbit_event:notify(queue_created,
                                          infos(?CREATION_EVENT_KEYS, State1)),
                      rabbit_event:if_enabled(StatsTimer,
@@ -387,17 +403,16 @@ deliver_msgs_to_consumers(Funs = {PredFun, DeliverFun}, FunAcc,
 deliver_from_queue_pred(IsEmpty, _State) ->
     not IsEmpty.
 
-deliver_from_queue_deliver(AckRequired, false,
-                           State = #q{backing_queue = BQ,
-                                      backing_queue_state = BQS}) ->
-    {{Message, IsDelivered, AckTag, Remaining}, BQS1} =
-        BQ:fetch(AckRequired, BQS),
-    {{Message, IsDelivered, AckTag}, 0 == Remaining,
-     State #q { backing_queue_state = BQS1 }}.
+deliver_from_queue_deliver(AckRequired, false, State) ->
+    {{Message, IsDelivered, AckTag, Remaining}, State1} =
+        fetch(AckRequired, State),
+    {{Message, IsDelivered, AckTag}, 0 == Remaining, State1}.
 
-run_message_queue(State = #q{backing_queue = BQ, backing_queue_state = BQS}) ->
+run_message_queue(State) ->
     Funs = {fun deliver_from_queue_pred/2,
             fun deliver_from_queue_deliver/3},
+    #q{backing_queue = BQ, backing_queue_state = BQS} = 
+        drop_expired_messages(State),
     IsEmpty = BQ:is_empty(BQS),
     {_IsEmpty1, State1} = deliver_msgs_to_consumers(Funs, IsEmpty, State),
     State1.
@@ -407,15 +422,17 @@ attempt_delivery(none, _ChPid, Message, State = #q{backing_queue = BQ}) ->
     DeliverFun =
         fun (AckRequired, false, State1 = #q{backing_queue_state = BQS}) ->
                 {AckTag, BQS1} =
-                    BQ:publish_delivered(AckRequired, Message, BQS),
+                    BQ:publish_delivered(AckRequired, Message, 
+                                         #msg_properties{}, BQS),
                 {{Message, false, AckTag}, true,
                  State1#q{backing_queue_state = BQS1}}
         end,
     deliver_msgs_to_consumers({ PredFun, DeliverFun }, false, State);
-attempt_delivery(Txn, ChPid, Message, State = #q{backing_queue = BQ,
+attempt_delivery(Txn, ChPid, Message, State = #q{backing_queue       = BQ, 
                                                  backing_queue_state = BQS}) ->
     record_current_channel_tx(ChPid, Txn),
-    {true, State#q{backing_queue_state = BQ:tx_publish(Txn, Message, BQS)}}.
+    {true, State#q{backing_queue_state = 
+                       BQ:tx_publish(Txn, Message, #msg_properties{}, BQS)}}.
 
 deliver_or_enqueue(Txn, ChPid, Message, State = #q{backing_queue = BQ}) ->
     case attempt_delivery(Txn, ChPid, Message, State) of
@@ -423,13 +440,27 @@ deliver_or_enqueue(Txn, ChPid, Message, State = #q{backing_queue = BQ}) ->
             {true, NewState};
         {false, NewState} ->
             %% Txn is none and no unblocked channels with consumers
-            BQS = BQ:publish(Message, State #q.backing_queue_state),
-            {false, NewState#q{backing_queue_state = BQS}}
+            BQS = BQ:publish(Message, 
+                             msg_properties(State), 
+                             State #q.backing_queue_state),
+            {false, ensure_ttl_timer(NewState#q{backing_queue_state = BQS})}
     end.
 
 requeue_and_run(AckTags, State = #q{backing_queue = BQ}) ->
     maybe_run_queue_via_backing_queue(
-      fun (BQS) -> BQ:requeue(AckTags, BQS) end, State).
+      fun (BQS) -> 
+              BQ:requeue(AckTags, reset_msg_expiry_fun(State), BQS)
+      end, State).
+
+fetch(AckRequired, State = #q{backing_queue_state = BQS, 
+                                  backing_queue = BQ}) ->
+    case BQ:fetch(AckRequired, BQS) of
+        {empty, BQS1} -> 
+            {empty, State#q{backing_queue_state = BQS1}};
+        {{Message, IsDelivered, AckTag, Remaining}, BQS1} ->
+            {{Message, IsDelivered, AckTag, Remaining}, 
+               State#q{backing_queue_state = BQS1}}
+    end.
 
 add_consumer(ChPid, Consumer, Queue) -> queue:in({ChPid, Consumer}, Queue).
 
@@ -526,8 +557,10 @@ maybe_run_queue_via_backing_queue(Fun, State = #q{backing_queue_state = BQS}) ->
 
 commit_transaction(Txn, From, ChPid, State = #q{backing_queue = BQ,
                                                 backing_queue_state = BQS}) ->
-    {AckTags, BQS1} =
-        BQ:tx_commit(Txn, fun () -> gen_server2:reply(From, ok) end, BQS),
+    {AckTags, BQS1} = BQ:tx_commit(Txn, 
+                                   fun () -> gen_server2:reply(From, ok) end, 
+                                   reset_msg_expiry_fun(State), 
+                                   BQS),
     %% ChPid must be known here because of the participant management
     %% by the channel.
     C = #cr{acktags = ChAckTags} = lookup_ch(ChPid),
@@ -546,6 +579,47 @@ rollback_transaction(Txn, ChPid, State = #q{backing_queue = BQ,
 subtract_acks(A, B) when is_list(B) ->
     lists:foldl(fun sets:del_element/2, A, B).
 
+reset_msg_expiry_fun(State) ->
+    fun(MsgProps) ->
+            MsgProps#msg_properties{expiry=calculate_msg_expiry(State)}
+    end.
+
+msg_properties(State) ->
+    #msg_properties{expiry = calculate_msg_expiry(State)}.
+
+calculate_msg_expiry(_State = #q{ttl = undefined}) ->
+    undefined;
+calculate_msg_expiry(_State = #q{ttl = Ttl}) ->
+    Now = timer:now_diff(now(), {0,0,0}),
+    Now + (Ttl * 1000).                 
+
+drop_expired_messages(State = #q{ttl = undefined}) ->
+    State;
+drop_expired_messages(State = #q{backing_queue_state = BQS, 
+                                  backing_queue = BQ}) ->
+    Now = timer:now_diff(now(), {0,0,0}),
+    BQS1 = BQ:dropwhile(
+             fun (_MsgProperties = #msg_properties{expiry=Expiry}) ->
+                     Now > Expiry
+             end, BQS),
+    ensure_ttl_timer(State #q{backing_queue_state = BQS1}).
+
+ensure_ttl_timer(State = #q{backing_queue       = BQ, 
+                            backing_queue_state = BQS, 
+                            ttl                 = Ttl, 
+                            ttl_timer_ref       = undefined}) 
+  when Ttl =/= undefined->
+    case BQ:is_empty(BQS) of
+        true ->
+            State;
+        false ->
+            State#q{ttl_timer_ref = 
+                        timer:send_after(Ttl, self(), drop_expired)}
+    end;
+ensure_ttl_timer(State) ->
+    State.
+            
+    
 infos(Items, State) -> [{Item, i(Item, State)} || Item <- Items].
 
 i(name,        #q{q = #amqqueue{name        = Name}})       -> Name;
@@ -670,6 +744,9 @@ handle_call({deliver_immediately, Txn, Message, ChPid}, _From, State) ->
     %% just all ready-to-consume queues get the message, with unready
     %% queues discarding the message?
     %%
+
+    %% we don't need an expiry here because messages are not being
+    %% enqueued, so we use an empty msg_properties.
     {Delivered, NewState} = attempt_delivery(Txn, ChPid, Message, State),
     reply(Delivered, NewState);
 
@@ -694,21 +771,21 @@ handle_call({notify_down, ChPid}, _From, State) ->
     end;
 
 handle_call({basic_get, ChPid, NoAck}, _From,
-            State = #q{q = #amqqueue{name = QName},
-                       backing_queue_state = BQS, backing_queue = BQ}) ->
+            State = #q{q = #amqqueue{name = QName}}) ->
     AckRequired = not NoAck,
     State1 = ensure_expiry_timer(State),
-    case BQ:fetch(AckRequired, BQS) of
-        {empty, BQS1} -> reply(empty, State1#q{backing_queue_state = BQS1});
-        {{Message, IsDelivered, AckTag, Remaining}, BQS1} ->
+    case fetch(AckRequired, drop_expired_messages(State1)) of
+        {empty, State2} -> 
+            reply(empty, State2);
+        {{Message, IsDelivered, AckTag, Remaining}, State2} ->
             case AckRequired of
-                true ->  C = #cr{acktags = ChAckTags} = ch_record(ChPid),
-                         store_ch_record(
+                true  ->  C = #cr{acktags = ChAckTags} = ch_record(ChPid),
+                          store_ch_record(
                            C#cr{acktags = sets:add_element(AckTag, ChAckTags)});
                 false -> ok
             end,
             Msg = {QName, self(), AckTag, IsDelivered, Message},
-            reply({ok, Remaining, Msg}, State1#q{backing_queue_state = BQS1})
+            reply({ok, Remaining, Msg}, State2)
     end;
 
 handle_call({basic_consume, NoAck, ChPid, LimiterPid,
@@ -949,6 +1026,9 @@ handle_info(timeout, State = #q{backing_queue = BQ}) ->
     noreply(maybe_run_queue_via_backing_queue(
               fun (BQS) -> BQ:idle_timeout(BQS) end, State));
 
+handle_info(drop_expired, State) ->
+    noreply(drop_expired_messages(State#q{ttl_timer_ref = undefined}));
+
 handle_info({'EXIT', _Pid, Reason}, State) ->
     {stop, Reason, State};
 
diff --git a/src/rabbit_backing_queue.erl b/src/rabbit_backing_queue.erl
index 2230c507..4f71c1a8 100644
--- a/src/rabbit_backing_queue.erl
+++ b/src/rabbit_backing_queue.erl
@@ -62,12 +62,16 @@ behaviour_info(callbacks) ->
      {purge, 1},
 
      %% Publish a message.
-     {publish, 2},
+     {publish, 3},
 
      %% Called for messages which have already been passed straight
      %% out to a client. The queue will be empty for these calls
      %% (i.e. saves the round trip through the backing queue).
-     {publish_delivered, 3},
+     {publish_delivered, 4},
+
+     %% Drop messages in the queue while the supplied predicate
+     %% returns true and return the new state.
+     {dropwhile, 2},
 
      %% Produce the next message.
      {fetch, 2},
@@ -77,7 +81,7 @@ behaviour_info(callbacks) ->
      {ack, 2},
 
      %% A publish, but in the context of a transaction.
-     {tx_publish, 3},
+     {tx_publish, 4},
 
      %% Acks, but in the context of a transaction.
      {tx_ack, 3},
@@ -89,11 +93,11 @@ behaviour_info(callbacks) ->
      %% Commit a transaction. The Fun passed in must be called once
      %% the messages have really been commited. This CPS permits the
      %% possibility of commit coalescing.
-     {tx_commit, 3},
+     {tx_commit, 4},
 
      %% Reinsert messages into the queue which have already been
      %% delivered and were pending acknowledgement.
-     {requeue, 2},
+     {requeue, 3},
 
      %% How long is my queue?
      {len, 1},
diff --git a/src/rabbit_invariable_queue.erl b/src/rabbit_invariable_queue.erl
index 4e0dad84..feb7c7e1 100644
--- a/src/rabbit_invariable_queue.erl
+++ b/src/rabbit_invariable_queue.erl
@@ -31,9 +31,9 @@
 
 -module(rabbit_invariable_queue).
 
--export([init/3, terminate/1, delete_and_terminate/1, purge/1, publish/2,
-         publish_delivered/3, fetch/2, ack/2, tx_publish/3, tx_ack/3,
-         tx_rollback/2, tx_commit/3, requeue/2, len/1, is_empty/1,
+-export([init/3, terminate/1, delete_and_terminate/1, purge/1, publish/3,
+         publish_delivered/4, fetch/2, ack/2, tx_publish/4, tx_ack/3,
+         dropwhile/2, tx_rollback/2, tx_commit/4, requeue/3, len/1, is_empty/1,
          set_ram_duration_target/2, ram_duration/1, needs_idle_timeout/1,
          idle_timeout/1, handle_pre_hibernate/1, status/1]).
 
@@ -89,40 +89,64 @@ purge(State = #iv_state { queue = Q, qname = QName, durable = IsDurable,
     %% We do not purge messages pending acks.
     {AckTags, PA} =
         rabbit_misc:queue_fold(
-          fun ({#basic_message { is_persistent = false }, _IsDelivered}, Acc) ->
+          fun ({#basic_message { is_persistent = false }, 
+                _MsgProps, _IsDelivered}, Acc) ->
                   Acc;
-              ({Msg = #basic_message { guid = Guid }, IsDelivered},
+              ({Msg = #basic_message { guid = Guid }, MsgProps, IsDelivered},
                {AckTagsN, PAN}) ->
                   ok = persist_delivery(QName, IsDurable, IsDelivered, Msg),
-                  {[Guid | AckTagsN], dict:store(Guid, Msg, PAN)}
+                  {[Guid | AckTagsN], store_ack(Msg, MsgProps, PAN)}
           end, {[], dict:new()}, Q),
     ok = persist_acks(QName, IsDurable, none, AckTags, PA),
     {Len, State #iv_state { len = 0, queue = queue:new() }}.
 
-publish(Msg, State = #iv_state { queue = Q, qname = QName, durable = IsDurable,
-                                 len = Len }) ->
-    ok = persist_message(QName, IsDurable, none, Msg),
-    State #iv_state { queue = queue:in({Msg, false}, Q), len = Len + 1 }.
+publish(Msg, MsgProps, State = #iv_state { queue   = Q, 
+                                           qname   = QName, 
+                                           durable = IsDurable,
+                                           len     = Len }) ->
+    ok = persist_message(QName, IsDurable, none, Msg, MsgProps),
+    Q1 = enqueue(Msg, MsgProps, false, Q),
+    State #iv_state { queue = Q1, len = Len + 1 }.
 
-publish_delivered(false, _Msg, State) ->
+publish_delivered(false, _Msg, _MsgProps, State) ->
     {blank_ack, State};
 publish_delivered(true, Msg = #basic_message { guid = Guid },
+                  MsgProps,
                   State = #iv_state { qname = QName, durable = IsDurable,
                                       len = 0, pending_ack = PA }) ->
-    ok = persist_message(QName, IsDurable, none, Msg),
+    ok = persist_message(QName, IsDurable, none, Msg, MsgProps),
     ok = persist_delivery(QName, IsDurable, false, Msg),
-    {Guid, State #iv_state { pending_ack = dict:store(Guid, Msg, PA) }}.
+    {Guid, State #iv_state { pending_ack = store_ack(Msg, MsgProps, PA) }}.
+
+dropwhile(_Pred, State = #iv_state { len = 0 }) ->
+    State;
+dropwhile(Pred, State = #iv_state { queue = Q }) ->
+    {{value, {Msg, MsgProps, IsDelivered}}, Q1} = queue:out(Q),
+    case Pred(MsgProps) of
+        true ->
+            {_, State1} = 
+                fetch_internal(false, Q1, Msg, MsgProps, IsDelivered, State),
+            dropwhile(Pred, State1);
+        false ->
+            State
+     end.
 
 fetch(_AckRequired, State = #iv_state { len = 0 }) ->
     {empty, State};
-fetch(AckRequired, State = #iv_state { len = Len, queue = Q, qname = QName,
-                                       durable = IsDurable,
-                                       pending_ack = PA }) ->
-    {{value, {Msg = #basic_message { guid = Guid }, IsDelivered}}, Q1} =
-        queue:out(Q),
+fetch(AckRequired, State = #iv_state { queue = Q }) ->
+    {{value, {Msg, MsgProps, IsDelivered}}, Q1} = queue:out(Q),
+    fetch_internal(AckRequired, Q1, Msg, MsgProps, IsDelivered, State).
+
+fetch_internal(AckRequired, Q1, 
+               Msg = #basic_message {guid = Guid}, 
+               MsgProps, IsDelivered,
+               State = #iv_state { len         = Len, 
+                                   qname       = QName,
+                                   durable     = IsDurable,
+                                   pending_ack = PA }) ->
     Len1 = Len - 1,
     ok = persist_delivery(QName, IsDurable, IsDelivered, Msg),
-    PA1 = dict:store(Guid, Msg, PA),
+    PA1 = store_ack(Msg, MsgProps, PA),
     {AckTag, PA2} = case AckRequired of
                         true  -> {Guid, PA1};
                         false -> ok = persist_acks(QName, IsDurable, none,
@@ -138,11 +162,11 @@ ack(AckTags, State = #iv_state { qname = QName, durable = IsDurable,
     PA1 = remove_acks(AckTags, PA),
     State #iv_state { pending_ack = PA1 }.
 
-tx_publish(Txn, Msg, State = #iv_state { qname = QName,
+tx_publish(Txn, Msg, MsgProps, State = #iv_state { qname = QName,
                                          durable = IsDurable }) ->
     Tx = #tx { pending_messages = Pubs } = lookup_tx(Txn),
-    store_tx(Txn, Tx #tx { pending_messages = [Msg | Pubs] }),
-    ok = persist_message(QName, IsDurable, Txn, Msg),
+    store_tx(Txn, Tx #tx { pending_messages = [{Msg, MsgProps} | Pubs] }),
+    ok = persist_message(QName, IsDurable, Txn, Msg, MsgProps),
     State.
 
 tx_ack(Txn, AckTags, State = #iv_state { qname = QName, durable = IsDurable,
@@ -159,8 +183,10 @@ tx_rollback(Txn, State = #iv_state { qname = QName }) ->
     erase_tx(Txn),
     {lists:flatten(AckTags), State}.
 
-tx_commit(Txn, Fun, State = #iv_state { qname = QName, pending_ack = PA,
-                                        queue = Q, len = Len }) ->
+tx_commit(Txn, Fun, MsgPropsFun, State = #iv_state { qname       = QName, 
+                                                     pending_ack = PA, 
+                                                     queue       = Q, 
+                                                     len         = Len }) ->
     #tx { pending_acks = AckTags, pending_messages = PubsRev } = lookup_tx(Txn),
     ok = do_if_persistent(fun rabbit_persister:commit_transaction/1,
                           Txn, QName),
@@ -168,13 +194,16 @@ tx_commit(Txn, Fun, State = #iv_state { qname = QName, pending_ack = PA,
     Fun(),
     AckTags1 = lists:flatten(AckTags),
     PA1 = remove_acks(AckTags1, PA),
-    {Q1, Len1} = lists:foldr(fun (Msg, {QN, LenN}) ->
-                                     {queue:in({Msg, false}, QN), LenN + 1}
+    {Q1, Len1} = lists:foldr(fun ({Msg, MsgProps}, {QN, LenN}) ->
+                                     {enqueue(Msg, MsgPropsFun(MsgProps), 
+                                              false, QN), 
+                                      LenN + 1}
                              end, {Q, Len}, PubsRev),
     {AckTags1, State #iv_state { pending_ack = PA1, queue = Q1, len = Len1 }}.
 
-requeue(AckTags, State = #iv_state { pending_ack = PA, queue = Q,
-                                     len = Len }) ->
+requeue(AckTags, MsgPropsFun, State = #iv_state { pending_ack = PA, 
+                                                  queue       = Q,
+                                                  len         = Len }) ->
     %% We don't need to touch the persister here - the persister will
     %% already have these messages published and delivered as
     %% necessary. The complication is that the persister's seq_id will
@@ -186,12 +215,17 @@ requeue(AckTags, State = #iv_state { pending_ack = PA, queue = Q,
     %% order to the last known state of our queue, prior to shutdown.
     {Q1, Len1} = lists:foldl(
                    fun (Guid, {QN, LenN}) ->
-                           {ok, Msg = #basic_message {}} = dict:find(Guid, PA),
-                           {queue:in({Msg, true}, QN), LenN + 1}
+                           {Msg = #basic_message {}, MsgProps}
+                               = dict:fetch(Guid, PA),
+                           {enqueue(Msg, MsgPropsFun(MsgProps), true, QN), 
+                            LenN + 1}
                    end, {Q, Len}, AckTags),
     PA1 = remove_acks(AckTags, PA),
     State #iv_state { pending_ack = PA1, queue = Q1, len = Len1 }.
 
+enqueue(Msg, MsgProps, IsDelivered, Q) ->
+    queue:in({Msg, MsgProps, IsDelivered}, Q).
+
 len(#iv_state { len = Len }) -> Len.
 
 is_empty(State) -> 0 == len(State).
@@ -212,6 +246,9 @@ status(_State) -> [].
 
 remove_acks(AckTags, PA) -> lists:foldl(fun dict:erase/2, PA, AckTags).
 
+store_ack(Msg = #basic_message { guid = Guid }, MsgProps, PA) ->
+    dict:store(Guid, {Msg, MsgProps}, PA).
+
 %%----------------------------------------------------------------------------
 
 lookup_tx(Txn) ->
@@ -243,14 +280,15 @@ do_if_persistent(F, Txn, QName) ->
 %%----------------------------------------------------------------------------
 
 persist_message(QName, true, Txn, Msg = #basic_message {
-                                    is_persistent = true }) ->
+                                    is_persistent = true }, MsgProps) ->
     Msg1 = Msg #basic_message {
              %% don't persist any recoverable decoded properties
              content = rabbit_binary_parser:clear_decoded_content(
                          Msg #basic_message.content)},
     persist_work(Txn, QName,
-                 [{publish, Msg1, {QName, Msg1 #basic_message.guid}}]);
-persist_message(_QName, _IsDurable, _Txn, _Msg) ->
+                 [{publish, Msg1, MsgProps, 
+                   {QName, Msg1 #basic_message.guid}}]);
+persist_message(_QName, _IsDurable, _Txn, _Msg, _MsgProps) ->
     ok.
 
 persist_delivery(QName, true, false, #basic_message { is_persistent = true,
@@ -263,7 +301,8 @@ persist_acks(QName, true, Txn, AckTags, PA) ->
     persist_work(Txn, QName,
                  [{ack, {QName, Guid}} || Guid <- AckTags,
                                           begin
-                                              {ok, Msg} = dict:find(Guid, PA),
+                                              {ok, {Msg, _MsgProps}} 
+                                                  = dict:find(Guid, PA),
                                               Msg #basic_message.is_persistent
                                           end]);
 persist_acks(_QName, _IsDurable, _Txn, _AckTags, _PA) ->
diff --git a/src/rabbit_persister.erl b/src/rabbit_persister.erl
index 66e5cf63..513b14df 100644
--- a/src/rabbit_persister.erl
+++ b/src/rabbit_persister.erl
@@ -69,7 +69,7 @@
 -type(pmsg() :: {rabbit_amqqueue:name(), pkey()}).
 
 -type(work_item() ::
-      {publish, rabbit_types:message(), pmsg()} |
+      {publish, rabbit_types:message(), rabbit_types:msg_properties(), pmsg()} |
       {deliver, pmsg()} |
       {ack, pmsg()}).
 
@@ -173,9 +173,10 @@ handle_call(force_snapshot, _From, State) ->
 handle_call({queue_content, QName}, _From,
             State = #pstate{snapshot = #psnapshot{messages = Messages,
                                                   queues   = Queues}}) ->
-    MatchSpec= [{{{QName,'$1'}, '$2', '$3'}, [], [{{'$3', '$1', '$2'}}]}],
-    do_reply([{ets:lookup_element(Messages, K, 2), D} ||
-                 {_, K, D} <- lists:sort(ets:select(Queues, MatchSpec))],
+    MatchSpec= [{{{QName,'$1'}, '$2', '$3', '$4'}, [], 
+                 [{{'$4', '$1', '$2', '$3'}}]}],
+    do_reply([{ets:lookup_element(Messages, K, 2), MP, D} ||
+                 {_, K, D, MP} <- lists:sort(ets:select(Queues, MatchSpec))],
              State);
 handle_call(_Request, _From, State) ->
     {noreply, State}.
@@ -243,9 +244,9 @@ log_work(CreateWorkUnit, MessageList,
            snapshot = Snapshot = #psnapshot{messages = Messages}}) ->
     Unit = CreateWorkUnit(
              rabbit_misc:map_in_order(
-               fun (M = {publish, Message, QK = {_QName, PKey}}) ->
+               fun (M = {publish, Message, MsgProps, QK = {_QName, PKey}}) ->
                        case ets:lookup(Messages, PKey) of
-                           [_] -> {tied, QK};
+                           [_] -> {tied, MsgProps, QK};
                            []  -> ets:insert(Messages, {PKey, Message}),
                                   M
                        end;
@@ -356,7 +357,8 @@ current_snapshot(_Snapshot = #psnapshot{transactions = Ts,
                                         next_seq_id  = NextSeqId}) ->
     %% Avoid infinite growth of the table by removing messages not
     %% bound to a queue anymore
-    PKeys = ets:foldl(fun ({{_QName, PKey}, _Delivered, _SeqId}, S) ->
+    PKeys = ets:foldl(fun ({{_QName, PKey}, _Delivered, 
+                            _MsgProps, _SeqId}, S) ->
                               sets:add_element(PKey, S)
                       end, sets:new(), Queues),
     prune_table(Messages, fun (Key) -> sets:is_element(Key, PKeys) end),
@@ -474,14 +476,14 @@ perform_work(MessageList, Messages, Queues, SeqId) ->
                         perform_work_item(Item, Messages, Queues, NextSeqId)
                 end, SeqId, MessageList).
 
-perform_work_item({publish, Message, QK = {_QName, PKey}},
+perform_work_item({publish, Message, MsgProps, QK = {_QName, PKey}},
                   Messages, Queues, NextSeqId) ->
     true = ets:insert(Messages, {PKey, Message}),
-    true = ets:insert(Queues, {QK, false, NextSeqId}),
+    true = ets:insert(Queues, {QK, false, MsgProps, NextSeqId}),
     NextSeqId + 1;
 
-perform_work_item({tied, QK}, _Messages, Queues, NextSeqId) ->
-    true = ets:insert(Queues, {QK, false, NextSeqId}),
+perform_work_item({tied, MsgProps, QK}, _Messages, Queues, NextSeqId) ->
+    true = ets:insert(Queues, {QK, false, MsgProps, NextSeqId}),
     NextSeqId + 1;
 
 perform_work_item({deliver, QK}, _Messages, Queues, NextSeqId) ->
diff --git a/src/rabbit_queue_index.erl b/src/rabbit_queue_index.erl
index 0b98290c..6568aa70 100644
--- a/src/rabbit_queue_index.erl
+++ b/src/rabbit_queue_index.erl
@@ -31,7 +31,7 @@
 
 -module(rabbit_queue_index).
 
--export([init/4, terminate/2, delete_and_terminate/1, publish/4,
+-export([init/4, terminate/2, delete_and_terminate/1, publish/5,
          deliver/2, ack/2, sync/2, flush/1, read/3,
          next_segment_boundary/1, bounds/1, recover/1]).
 
@@ -98,7 +98,7 @@
 %% and seeding the message store on start up.
 %%
 %% Note that in general, the representation of a message's state as
-%% the tuple: {('no_pub'|{Guid, IsPersistent}), ('del'|'no_del'),
+%% the tuple: {('no_pub'|{Guid, MsgProperties, IsPersistent}), ('del'|'no_del'),
 %% ('ack'|'no_ack')} is richer than strictly necessary for most
 %% operations. However, for startup, and to ensure the safe and
 %% correct combination of journal entries with entries read from the
@@ -141,14 +141,19 @@
 -define(REL_SEQ_ONLY_ENTRY_LENGTH_BYTES, 2).
 
 %% publish record is binary 1 followed by a bit for is_persistent,
-%% then 14 bits of rel seq id, and 128 bits of md5sum msg id
+%% then 14 bits of rel seq id, 64 bits for message expiry and 128 bits 
+%% of md5sum msg id
 -define(PUBLISH_PREFIX, 1).
 -define(PUBLISH_PREFIX_BITS, 1).
 
+-define(EXPIRY_BYTES, 8).
+-define(EXPIRY_BITS, (?EXPIRY_BYTES * 8)).
+-define(NO_EXPIRY, 0).
+
 -define(GUID_BYTES, 16). %% md5sum is 128 bit or 16 bytes
 -define(GUID_BITS, (?GUID_BYTES * 8)).
-%% 16 bytes for md5sum + 2 for seq, bits and prefix
--define(PUBLISH_RECORD_LENGTH_BYTES, ?GUID_BYTES + 2).
+%% 16 bytes for md5sum + 8 for expiry + 2 for seq, bits and prefix
+-define(PUBLISH_RECORD_LENGTH_BYTES, ?GUID_BYTES + ?EXPIRY_BYTES + 2).
 
 %% 1 publish, 1 deliver, 1 ack per msg
 -define(SEGMENT_TOTAL_SIZE, ?SEGMENT_ENTRY_COUNT *
@@ -157,7 +162,7 @@
 
 %% ---- misc ----
 
--define(PUB, {_, _}). %% {Guid, IsPersistent}
+-define(PUB, {_, _, _}). %% {Guid, MsgProperties, IsPersistent}
 
 -define(READ_MODE, [binary, raw, read]).
 -define(READ_AHEAD_MODE, [{read_ahead, ?SEGMENT_TOTAL_SIZE} | ?READ_MODE]).
@@ -200,15 +205,16 @@
              {'undefined' | non_neg_integer(), [any()], qistate()}).
 -spec(terminate/2 :: ([any()], qistate()) -> qistate()).
 -spec(delete_and_terminate/1 :: (qistate()) -> qistate()).
--spec(publish/4 :: (rabbit_guid:guid(), seq_id(), boolean(), qistate()) ->
-                        qistate()).
+-spec(publish/5 :: (rabbit_guid:guid(), seq_id(), rabbit_types:msg_properties(),
+                       boolean(), qistate()) -> qistate()).
 -spec(deliver/2 :: ([seq_id()], qistate()) -> qistate()).
 -spec(ack/2 :: ([seq_id()], qistate()) -> qistate()).
 -spec(sync/2 :: ([seq_id()], qistate()) -> qistate()).
 -spec(flush/1 :: (qistate()) -> qistate()).
 -spec(read/3 :: (seq_id(), seq_id(), qistate()) ->
-                     {[{rabbit_guid:guid(), seq_id(), boolean(), boolean()}],
-                      qistate()}).
+                     {[{rabbit_guid:guid(), seq_id(), 
+                        rabbit_types:msg_properties(), 
+                        boolean(), boolean()}], qistate()}).
 -spec(next_segment_boundary/1 :: (seq_id()) -> seq_id()).
 -spec(bounds/1 :: (qistate()) ->
              {non_neg_integer(), non_neg_integer(), qistate()}).
@@ -252,15 +258,19 @@ delete_and_terminate(State) ->
     ok = rabbit_misc:recursive_delete([Dir]),
     State1.
 
-publish(Guid, SeqId, IsPersistent, State) when is_binary(Guid) ->
+publish(Guid, SeqId, MsgProperties, IsPersistent, State) 
+  when is_binary(Guid) ->
     ?GUID_BYTES = size(Guid),
     {JournalHdl, State1} = get_journal_handle(State),
     ok = file_handle_cache:append(
            JournalHdl, [<<(case IsPersistent of
                                true  -> ?PUB_PERSIST_JPREFIX;
                                false -> ?PUB_TRANS_JPREFIX
-                           end):?JPREFIX_BITS, SeqId:?SEQ_BITS>>, Guid]),
-    maybe_flush_journal(add_to_journal(SeqId, {Guid, IsPersistent}, State1)).
+                           end):?JPREFIX_BITS, 
+                          SeqId:?SEQ_BITS>>,
+                          create_pub_record_body(Guid, MsgProperties)]),
+    maybe_flush_journal(
+      add_to_journal(SeqId, {Guid, MsgProperties, IsPersistent}, State1)).
 
 deliver(SeqIds, State) ->
     deliver_or_ack(del, SeqIds, State).
@@ -289,7 +299,7 @@ flush(State)                                -> flush_journal(State).
 
 read(StartEnd, StartEnd, State) ->
     {[], State};
-read(Start, End, State = #qistate { segments = Segments,
+read(Start, End, State =  #qistate { segments = Segments,
                                     dir = Dir }) when Start =< End ->
     %% Start is inclusive, End is exclusive.
     LowerB = {StartSeg, _StartRelSeq} = seq_id_to_seg_and_rel_seq_id(Start),
@@ -453,7 +463,9 @@ recover_segment(ContainsCheckFun, CleanShutdown,
     {SegEntries1, UnackedCountDelta} =
         segment_plus_journal(SegEntries, JEntries),
     array:sparse_foldl(
-      fun (RelSeq, {{Guid, _IsPersistent}, Del, no_ack}, Segment1) ->
+      fun (RelSeq, 
+           {{Guid, _MsgProperties, _IsPersistent}, Del, no_ack}, 
+           Segment1) ->
               recover_message(ContainsCheckFun(Guid), CleanShutdown,
                               Del, RelSeq, Segment1)
       end,
@@ -506,7 +518,9 @@ queue_index_walker_reader(QueueName, Gatherer) ->
     State = #qistate { segments = Segments, dir = Dir } =
         recover_journal(blank_state(QueueName)),
     [ok = segment_entries_foldr(
-            fun (_RelSeq, {{Guid, true}, _IsDelivered, no_ack}, ok) ->
+            fun (_RelSeq, 
+                 {{Guid, _MsgProps, true}, _IsDelivered, no_ack}, 
+                 ok) ->
                     gatherer:in(Gatherer, {Guid, 1});
                 (_RelSeq, _Value, Acc) ->
                     Acc
@@ -516,6 +530,34 @@ queue_index_walker_reader(QueueName, Gatherer) ->
     ok = gatherer:finish(Gatherer).
 
 %%----------------------------------------------------------------------------
+%% expiry/binary manipulation
+%%----------------------------------------------------------------------------
+
+create_pub_record_body(Guid, #msg_properties{expiry = Expiry}) ->
+    [Guid, expiry_to_binary(Expiry)].
+
+expiry_to_binary(undefined) ->
+    <<?NO_EXPIRY:?EXPIRY_BITS>>;
+expiry_to_binary(Expiry) ->
+    <<Expiry:?EXPIRY_BITS>>.
+
+read_pub_record_body(Hdl) ->
+    case file_handle_cache:read(Hdl, ?GUID_BYTES + ?EXPIRY_BYTES) of
+        {ok, Bin} ->
+            %% work around for binary data fragmentation. See
+            %% rabbit_msg_file:read_next/2
+            <<GuidNum:?GUID_BITS, Expiry:?EXPIRY_BITS>> = Bin,
+            <<Guid:?GUID_BYTES/binary>> = <<GuidNum:?GUID_BITS>>,
+            Exp = case Expiry of
+                      ?NO_EXPIRY -> undefined;
+                      X          -> X
+                  end,
+            {Guid, #msg_properties{expiry = Exp}};
+        Error ->
+            Error
+    end.
+            
+%%----------------------------------------------------------------------------
 %% journal manipulation
 %%----------------------------------------------------------------------------
 
@@ -636,14 +678,9 @@ load_journal_entries(State = #qistate { journal_handle = Hdl }) ->
                 ?ACK_JPREFIX ->
                     load_journal_entries(add_to_journal(SeqId, ack, State));
                 _ ->
-                    case file_handle_cache:read(Hdl, ?GUID_BYTES) of
-                        {ok, <<GuidNum:?GUID_BITS>>} ->
-                            %% work around for binary data
-                            %% fragmentation. See
-                            %% rabbit_msg_file:read_next/2
-                            <<Guid:?GUID_BYTES/binary>> =
-                                <<GuidNum:?GUID_BITS>>,
-                            Publish = {Guid, case Prefix of
+                    case read_pub_record_body(Hdl) of
+                        {Guid, MsgProperties} ->
+                            Publish = {Guid, MsgProperties, case Prefix of
                                                  ?PUB_PERSIST_JPREFIX -> true;
                                                  ?PUB_TRANS_JPREFIX   -> false
                                              end},
@@ -744,11 +781,12 @@ write_entry_to_segment(RelSeq, {Pub, Del, Ack}, Hdl) ->
     ok = case Pub of
              no_pub ->
                  ok;
-             {Guid, IsPersistent} ->
+             {Guid, MsgProperties, IsPersistent} ->
                  file_handle_cache:append(
                    Hdl, [<<?PUBLISH_PREFIX:?PUBLISH_PREFIX_BITS,
                           (bool_to_int(IsPersistent)):1,
-                          RelSeq:?REL_SEQ_BITS>>, Guid])
+                          RelSeq:?REL_SEQ_BITS>>,
+                          create_pub_record_body(Guid, MsgProperties)])
          end,
     ok = case {Del, Ack} of
              {no_del, no_ack} ->
@@ -768,10 +806,12 @@ read_bounded_segment(Seg, {StartSeg, StartRelSeq}, {EndSeg, EndRelSeq},
                      {Messages, Segments}, Dir) ->
     Segment = segment_find_or_new(Seg, Dir, Segments),
     {segment_entries_foldr(
-       fun (RelSeq, {{Guid, IsPersistent}, IsDelivered, no_ack}, Acc)
+       fun (RelSeq, 
+            {{Guid, MsgProperties, IsPersistent}, IsDelivered, no_ack}, 
+            Acc)
              when (Seg > StartSeg orelse StartRelSeq =< RelSeq) andalso
                   (Seg < EndSeg   orelse EndRelSeq   >= RelSeq) ->
-               [ {Guid, reconstruct_seq_id(StartSeg, RelSeq),
+               [ {Guid, reconstruct_seq_id(StartSeg, RelSeq), MsgProperties,
                   IsPersistent, IsDelivered == del} | Acc ];
            (_RelSeq, _Value, Acc) ->
                Acc
@@ -801,10 +841,8 @@ load_segment_entries(KeepAcked, Hdl, SegEntries, UnackedCount) ->
     case file_handle_cache:read(Hdl, ?REL_SEQ_ONLY_ENTRY_LENGTH_BYTES) of
         {ok, <<?PUBLISH_PREFIX:?PUBLISH_PREFIX_BITS,
               IsPersistentNum:1, RelSeq:?REL_SEQ_BITS>>} ->
-            %% because we specify /binary, and binaries are complete
-            %% bytes, the size spec is in bytes, not bits.
-            {ok, Guid} = file_handle_cache:read(Hdl, ?GUID_BYTES),
-            Obj = {{Guid, 1 == IsPersistentNum}, no_del, no_ack},
+            {Guid, MsgProperties} = read_pub_record_body(Hdl),
+            Obj = {{Guid, MsgProperties, 1 == IsPersistentNum}, no_del, no_ack},
             SegEntries1 = array:set(RelSeq, Obj, SegEntries),
             load_segment_entries(KeepAcked, Hdl, SegEntries1,
                                  UnackedCount + 1);
diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl
index a72656b7..430a79d9 100644
--- a/src/rabbit_tests.erl
+++ b/src/rabbit_tests.erl
@@ -1414,6 +1414,7 @@ test_backing_queue() ->
             application:set_env(rabbit, msg_store_file_size_limit,
                                 FileSizeLimit, infinity),
             passed = test_queue_index(),
+            passed = test_queue_index_props(),
             passed = test_variable_queue(),
             passed = test_queue_recover(),
             application:set_env(rabbit, queue_index_max_journal_entries,
@@ -1639,7 +1640,7 @@ queue_index_publish(SeqIds, Persistent, Qi) ->
           fun (SeqId, {QiN, SeqIdsGuidsAcc, MSCStateN}) ->
                   Guid = rabbit_guid:guid(),
                   QiM = rabbit_queue_index:publish(
-                          Guid, SeqId, Persistent, QiN),
+                          Guid, SeqId, #msg_properties{}, Persistent, QiN),
                   {ok, MSCStateM} = rabbit_msg_store:write(MsgStore, Guid,
                                                            Guid, MSCStateN),
                   {QiM, [{SeqId, Guid} | SeqIdsGuidsAcc], MSCStateM}
@@ -1651,12 +1652,28 @@ queue_index_publish(SeqIds, Persistent, Qi) ->
 verify_read_with_published(_Delivered, _Persistent, [], _) ->
     ok;
 verify_read_with_published(Delivered, Persistent,
-                           [{Guid, SeqId, Persistent, Delivered}|Read],
+                           [{Guid, SeqId, _Props, Persistent, Delivered}|Read],
                            [{SeqId, Guid}|Published]) ->
     verify_read_with_published(Delivered, Persistent, Read, Published);
 verify_read_with_published(_Delivered, _Persistent, _Read, _Published) ->
     ko.
 
+test_queue_index_props() ->
+    with_empty_test_queue(
+      fun(Qi0) ->
+              Guid = rabbit_guid:guid(),
+              Props = #msg_properties{expiry=12345},
+              Qi1 = rabbit_queue_index:publish(Guid, 1, Props, true, Qi0),
+              {[{Guid, 1, Props, _, _}], Qi2} = 
+                  rabbit_queue_index:read(1, 2, Qi1),
+              Qi2
+      end),
+
+    ok = rabbit_variable_queue:stop(),
+    ok = rabbit_variable_queue:start([]),
+
+    passed.
+
 test_queue_index() ->
     SegmentSize = rabbit_queue_index:next_segment_boundary(0),
     TwoSegs = SegmentSize + SegmentSize,
@@ -1786,11 +1803,12 @@ variable_queue_publish(IsPersistent, Count, VQ) ->
       fun (_N, VQN) ->
               rabbit_variable_queue:publish(
                 rabbit_basic:message(
-                  rabbit_misc:r(<<>>, exchange, <<>>),
+                  rabbit_misc:r(<<>>, exchange, <<>>), 
                   <<>>, #'P_basic'{delivery_mode = case IsPersistent of
                                                        true  -> 2;
                                                        false -> 1
-                                                   end}, <<>>), VQN)
+                                                   end}, <<>>), 
+                #msg_properties{}, VQN)
       end, VQ, lists:seq(1, Count)).
 
 variable_queue_fetch(Count, IsPersistent, IsDelivered, Len, VQ) ->
@@ -1824,9 +1842,41 @@ test_variable_queue() ->
         F <- [fun test_variable_queue_dynamic_duration_change/1,
               fun test_variable_queue_partial_segments_delta_thing/1,
               fun test_variable_queue_all_the_bits_not_covered_elsewhere1/1,
-              fun test_variable_queue_all_the_bits_not_covered_elsewhere2/1]],
+              fun test_variable_queue_all_the_bits_not_covered_elsewhere2/1,
+              fun test_dropwhile/1]],
     passed.
 
+test_dropwhile(VQ0) ->
+    Count = 10,
+
+    %% add messages with sequential expiry
+    VQ1 = lists:foldl(
+      fun (N, VQN) ->
+              rabbit_variable_queue:publish(
+                rabbit_basic:message(
+                  rabbit_misc:r(<<>>, exchange, <<>>), 
+                  <<>>, #'P_basic'{}, <<>>), 
+                #msg_properties{expiry = N}, VQN)
+      end, VQ0, lists:seq(1, Count)),
+
+    %% drop the first 5 messages
+    VQ2 = rabbit_variable_queue:dropwhile(
+      fun(#msg_properties { expiry = Expiry }) ->
+              Expiry =< 5
+      end, VQ1),
+
+    %% fetch five now
+    VQ3 = lists:foldl(fun (_N, VQN) ->
+                                   {{#basic_message{}, _, _, _}, VQM} =
+                                       rabbit_variable_queue:fetch(false, VQN),
+                                   VQM
+                           end, VQ2, lists:seq(1, 5)),
+
+    %% should be empty now
+    {empty, VQ4} = rabbit_variable_queue:fetch(false, VQ3),
+
+    VQ4.
+    
 test_variable_queue_dynamic_duration_change(VQ0) ->
     SegmentSize = rabbit_queue_index:next_segment_boundary(0),
 
@@ -1837,6 +1887,7 @@ test_variable_queue_dynamic_duration_change(VQ0) ->
     %% squeeze and relax queue
     Churn = Len div 32,
     VQ2 = publish_fetch_and_ack(Churn, Len, VQ1),
+
     {Duration, VQ3} = rabbit_variable_queue:ram_duration(VQ2),
     VQ7 = lists:foldl(
             fun (Duration1, VQ4) ->
@@ -1859,7 +1910,8 @@ publish_fetch_and_ack(0, _Len, VQ0) ->
     VQ0;
 publish_fetch_and_ack(N, Len, VQ0) ->
     VQ1 = variable_queue_publish(false, 1, VQ0),
-    {{_Msg, false, AckTag, Len}, VQ2} = rabbit_variable_queue:fetch(true, VQ1),
+    {{_Msg, false, AckTag, Len}, VQ2} = 
+        rabbit_variable_queue:fetch(true, VQ1),
     publish_fetch_and_ack(N-1, Len, rabbit_variable_queue:ack([AckTag], VQ2)).
 
 test_variable_queue_partial_segments_delta_thing(VQ0) ->
@@ -1935,7 +1987,7 @@ test_variable_queue_all_the_bits_not_covered_elsewhere2(VQ0) ->
     VQ1 = rabbit_variable_queue:set_ram_duration_target(0, VQ0),
     VQ2 = variable_queue_publish(false, 4, VQ1),
     {VQ3, AckTags} = variable_queue_fetch(2, false, false, 4, VQ2),
-    VQ4 = rabbit_variable_queue:requeue(AckTags, VQ3),
+    VQ4 = rabbit_variable_queue:requeue(AckTags, fun(X) -> X end, VQ3),
     VQ5 = rabbit_variable_queue:idle_timeout(VQ4),
     _VQ6 = rabbit_variable_queue:terminate(VQ5),
     VQ7 = rabbit_variable_queue:init(test_queue(), true, true),
diff --git a/src/rabbit_types.erl b/src/rabbit_types.erl
index 0b6a15ec..35f74da3 100644
--- a/src/rabbit_types.erl
+++ b/src/rabbit_types.erl
@@ -41,7 +41,8 @@
               amqp_error/0, r/1, r2/2, r3/3, listener/0,
               binding/0, amqqueue/0, exchange/0, connection/0, protocol/0,
               user/0, ok/1, error/1, ok_or_error/1, ok_or_error2/2,
-              ok_pid_or_error/0, channel_exit/0, connection_exit/0]).
+              ok_pid_or_error/0, channel_exit/0, connection_exit/0,
+              msg_properties/0]).
 
 -type(channel_exit() :: no_return()).
 -type(connection_exit() :: no_return()).
@@ -86,6 +87,8 @@
                 txn       :: maybe(txn()),
                 sender    :: pid(),
                 message   :: message()}).
+-type(msg_properties() ::
+        #msg_properties{expiry :: pos_integer()}).
 
 %% this is really an abstract type, but dialyzer does not support them
 -type(txn() :: rabbit_guid:guid()).
diff --git a/src/rabbit_variable_queue.erl b/src/rabbit_variable_queue.erl
index 30d3a8ae..7d584026 100644
--- a/src/rabbit_variable_queue.erl
+++ b/src/rabbit_variable_queue.erl
@@ -32,9 +32,9 @@
 -module(rabbit_variable_queue).
 
 -export([init/3, terminate/1, delete_and_terminate/1,
-         purge/1, publish/2, publish_delivered/3, fetch/2, ack/2,
-         tx_publish/3, tx_ack/3, tx_rollback/2, tx_commit/3,
-         requeue/2, len/1, is_empty/1,
+         purge/1, publish/3, publish_delivered/4, fetch/2, ack/2,
+         tx_publish/4, tx_ack/3, tx_rollback/2, tx_commit/4,
+         requeue/3, len/1, is_empty/1, dropwhile/2,
          set_ram_duration_target/2, ram_duration/1,
          needs_idle_timeout/1, idle_timeout/1, handle_pre_hibernate/1,
          status/1]).
@@ -248,7 +248,8 @@
           is_persistent,
           is_delivered,
           msg_on_disk,
-          index_on_disk
+          index_on_disk,
+          msg_properties
          }).
 
 -record(delta,
@@ -490,13 +491,14 @@ purge(State = #vqstate { q4 = Q4, index_state = IndexState, len = Len }) ->
                               ram_index_count  = 0,
                               persistent_count = 0 })}.
 
-publish(Msg, State) ->
-    {_SeqId, State1} = publish(Msg, false, false, State),
+publish(Msg, MsgProperties, State) ->
+    {_SeqId, State1} = publish(Msg, MsgProperties, false, false, State),
     a(reduce_memory_use(State1)).
 
-publish_delivered(false, _Msg, State = #vqstate { len = 0 }) ->
+publish_delivered(false, _Msg, _MsgProps, State = #vqstate { len = 0 }) ->
     {blank_ack, a(State)};
 publish_delivered(true, Msg = #basic_message { is_persistent = IsPersistent },
+                  MsgProps,
                   State = #vqstate { len               = 0,
                                      next_seq_id       = SeqId,
                                      out_counter       = OutCount,
@@ -505,7 +507,7 @@ publish_delivered(true, Msg = #basic_message { is_persistent = IsPersistent },
                                      pending_ack       = PA,
                                      durable           = IsDurable }) ->
     IsPersistent1 = IsDurable andalso IsPersistent,
-    MsgStatus = (msg_status(IsPersistent1, SeqId, Msg))
+    MsgStatus = (msg_status(IsPersistent1, SeqId, Msg, MsgProps))
         #msg_status { is_delivered = true },
     {MsgStatus1, State1} = maybe_write_to_disk(false, false, MsgStatus, State),
     PA1 = record_pending_ack(m(MsgStatus1), PA),
@@ -515,76 +517,128 @@ publish_delivered(true, Msg = #basic_message { is_persistent = IsPersistent },
                                 in_counter        = InCount  + 1,
                                 persistent_count  = PCount1,
                                 pending_ack       = PA1 })}.
+              
+dropwhile(Pred, State) ->
+    case internal_queue_out(
+           fun(MsgStatus = #msg_status { msg_properties = MsgProps },
+               State1) ->
+                   case Pred(MsgProps) of
+                       true ->
+                           {_, State2} = internal_fetch(false, 
+                                                        MsgStatus, State1),
+                           dropwhile(Pred, State2); 
+                       false ->
+                           %% message needs to go back into Q4 (or
+                           %% maybe go in for the first time if it was
+                           %% loaded from Q3). Also the msg contents
+                           %% might not be in RAM, so read them in now
+                           {MsgStatus1, State2 = #vqstate { q4 = Q4 }} = 
+                               read_msg(MsgStatus, State1),
+                           State2 #vqstate {q4 = queue:in_r(MsgStatus1, Q4)}
+                   end
+           end, State) of
+        {empty, StateR} -> StateR;
+        StateR -> StateR
+    end.
+
+fetch(AckRequired, State) ->
+    internal_queue_out(
+      fun(MsgStatus, State1) -> 
+              %% it's possible that the message wasn't read from disk
+              %% at this point, so read it in.
+              {MsgStatus1, State2} = read_msg(MsgStatus, State1),
+              internal_fetch(AckRequired, MsgStatus1, State2) 
+      end, State).
 
-fetch(AckRequired, State = #vqstate { q4               = Q4,
-                                      ram_msg_count    = RamMsgCount,
-                                      out_counter      = OutCount,
-                                      index_state      = IndexState,
-                                      len              = Len,
-                                      persistent_count = PCount,
-                                      pending_ack      = PA }) ->
+internal_queue_out(Fun, State = #vqstate { q4 = Q4 }) ->
     case queue:out(Q4) of
         {empty, _Q4} ->
-            case fetch_from_q3_to_q4(State) of
-                {empty, State1} = Result -> a(State1), Result;
-                {loaded, State1}         -> fetch(AckRequired, State1)
+            case fetch_from_q3(State) of
+                {empty, State1} = Result      -> a(State1), Result;
+                {loaded, {MsgStatus, State1}} -> Fun(MsgStatus, State1)
             end;
-        {{value, MsgStatus = #msg_status {
-                   msg = Msg, guid = Guid, seq_id = SeqId,
-                   is_persistent = IsPersistent, is_delivered = IsDelivered,
-                   msg_on_disk = MsgOnDisk, index_on_disk = IndexOnDisk }},
-         Q4a} ->
-
-            %% 1. Mark it delivered if necessary
-            IndexState1 = maybe_write_delivered(
-                            IndexOnDisk andalso not IsDelivered,
-                            SeqId, IndexState),
-
-            %% 2. Remove from msg_store and queue index, if necessary
-            MsgStore = find_msg_store(IsPersistent),
-            Rem = fun () -> ok = rabbit_msg_store:remove(MsgStore, [Guid]) end,
-            Ack = fun () -> rabbit_queue_index:ack([SeqId], IndexState1) end,
-            IndexState2 =
-                case {AckRequired, MsgOnDisk, IndexOnDisk, IsPersistent} of
-                    {false, true, false,     _} -> Rem(), IndexState1;
-                    {false, true,  true,     _} -> Rem(), Ack();
-                    { true, true,  true, false} -> Ack();
-                    _                           -> IndexState1
-                end,
-
-            %% 3. If an ack is required, add something sensible to PA
-            {AckTag, PA1} = case AckRequired of
-                                true  -> PA2 = record_pending_ack(
-                                                 MsgStatus #msg_status {
-                                                   is_delivered = true }, PA),
-                                         {SeqId, PA2};
-                                false -> {blank_ack, PA}
-                            end,
-
-            PCount1 = PCount - one_if(IsPersistent andalso not AckRequired),
-            Len1 = Len - 1,
-            {{Msg, IsDelivered, AckTag, Len1},
-             a(State #vqstate { q4               = Q4a,
-                                ram_msg_count    = RamMsgCount - 1,
-                                out_counter      = OutCount + 1,
-                                index_state      = IndexState2,
-                                len              = Len1,
-                                persistent_count = PCount1,
-                                pending_ack      = PA1 })}
+        {{value, MsgStatus}, Q4a} ->
+            Fun(MsgStatus, State #vqstate { q4 = Q4a })
     end.
 
+read_msg(MsgStatus = #msg_status { msg           = undefined, 
+                                   guid          = Guid, 
+                                   index_on_disk = IndexOnDisk,
+                                   is_persistent = IsPersistent }, 
+         State = #vqstate { ram_msg_count    = RamMsgCount, 
+                            ram_index_count  = RamIndexCount, 
+                            msg_store_clients = MSCState}) ->
+    {{ok, Msg = #basic_message {}}, MSCState1} = 
+        read_from_msg_store(MSCState, IsPersistent, Guid),
+
+    RamIndexCount1 = RamIndexCount - one_if(not IndexOnDisk),
+    true = RamIndexCount1 >= 0, %% ASSERTION
+
+    {MsgStatus #msg_status { msg = Msg }, 
+     State #vqstate { ram_msg_count     = RamMsgCount + 1,
+                      ram_index_count   = RamIndexCount1,
+                      msg_store_clients = MSCState1 }};
+read_msg(MsgStatus, State) ->
+    {MsgStatus, State}.
+
+internal_fetch(AckRequired,
+             MsgStatus = #msg_status {
+               msg = Msg, guid = Guid, seq_id = SeqId, 
+               is_persistent = IsPersistent, is_delivered = IsDelivered,
+               msg_on_disk = MsgOnDisk, index_on_disk = IndexOnDisk }, 
+             State = #vqstate { 
+               ram_msg_count = RamMsgCount, out_counter = OutCount, 
+               index_state = IndexState, len = Len, persistent_count = PCount, 
+               pending_ack = PA }) ->
+    %% 1. Mark it delivered if necessary
+    IndexState1 = maybe_write_delivered(
+                    IndexOnDisk andalso not IsDelivered,
+                    SeqId, IndexState),
+
+    %% 2. Remove from msg_store and queue index, if necessary
+    MsgStore = find_msg_store(IsPersistent),
+    Rem = fun () -> ok = rabbit_msg_store:remove(MsgStore, [Guid]) end,
+    Ack = fun () -> rabbit_queue_index:ack([SeqId], IndexState1) end,
+    IndexState2 =
+        case {AckRequired, MsgOnDisk, IndexOnDisk, IsPersistent} of
+            {false, true, false,     _} -> Rem(), IndexState1;
+            {false, true,  true,     _} -> Rem(), Ack();
+            { true, true,  true, false} -> Ack();
+            _                           -> IndexState1
+        end,
+
+    %% 3. If an ack is required, add something sensible to PA
+    {AckTag, PA1} = case AckRequired of
+                        true  -> PA2 = record_pending_ack(
+                                         MsgStatus #msg_status {
+                                           is_delivered = true }, PA),
+                                 {SeqId, PA2};
+                        false -> {blank_ack, PA}
+                    end,
+
+    PCount1 = PCount - one_if(IsPersistent andalso not AckRequired),
+    Len1 = Len - 1,
+    {{Msg, IsDelivered, AckTag, Len1},
+     a(State #vqstate { ram_msg_count    = RamMsgCount - 1,
+                        out_counter      = OutCount + 1,
+                        index_state      = IndexState2,
+                        len              = Len1,
+                        persistent_count = PCount1,
+                        pending_ack      = PA1 })}.
+    
 ack(AckTags, State) ->
     a(ack(fun rabbit_msg_store:remove/2,
           fun (_AckEntry, State1) -> State1 end,
           AckTags, State)).
 
 tx_publish(Txn, Msg = #basic_message { is_persistent = IsPersistent },
+           MsgProperties,
            State = #vqstate { durable           = IsDurable,
                               msg_store_clients = MSCState }) ->
     Tx = #tx { pending_messages = Pubs } = lookup_tx(Txn),
-    store_tx(Txn, Tx #tx { pending_messages = [Msg | Pubs] }),
+    store_tx(Txn, Tx #tx { pending_messages = [{Msg, MsgProperties} | Pubs] }),
     a(case IsPersistent andalso IsDurable of
-          true  -> MsgStatus = msg_status(true, undefined, Msg),
+          true  -> MsgStatus = msg_status(true, undefined, Msg, MsgProperties),
                    {#msg_status { msg_on_disk = true }, MSCState1} =
                        maybe_write_msg_to_disk(false, MsgStatus, MSCState),
                    State #vqstate { msg_store_clients = MSCState1 };
@@ -606,36 +660,39 @@ tx_rollback(Txn, State = #vqstate { durable = IsDurable }) ->
          end,
     {lists:append(AckTags), a(State)}.
 
-tx_commit(Txn, Fun, State = #vqstate { durable = IsDurable }) ->
+tx_commit(Txn, Fun, MsgPropsFun, State = #vqstate { durable = IsDurable }) ->
     #tx { pending_acks = AckTags, pending_messages = Pubs } = lookup_tx(Txn),
     erase_tx(Txn),
-    PubsOrdered = lists:reverse(Pubs),
     AckTags1 = lists:append(AckTags),
-    PersistentGuids = persistent_guids(PubsOrdered),
+    PersistentGuids = persistent_guids(Pubs),
     HasPersistentPubs = PersistentGuids =/= [],
     {AckTags1,
      a(case IsDurable andalso HasPersistentPubs of
            true  -> ok = rabbit_msg_store:sync(
                            ?PERSISTENT_MSG_STORE, PersistentGuids,
-                           msg_store_callback(PersistentGuids,
-                                              PubsOrdered, AckTags1, Fun)),
+                           msg_store_callback(PersistentGuids,Pubs, AckTags1, 
+                                              Fun, MsgPropsFun)),
                     State;
-           false -> tx_commit_post_msg_store(
-                      HasPersistentPubs, PubsOrdered, AckTags1, Fun, State)
+           false -> tx_commit_post_msg_store(HasPersistentPubs, Pubs, AckTags1, 
+                                             Fun, MsgPropsFun, State)
        end)}.
 
-requeue(AckTags, State) ->
+requeue(AckTags, MsgPropsFun, State) ->
     a(reduce_memory_use(
         ack(fun rabbit_msg_store:release/2,
-            fun (#msg_status { msg = Msg }, State1) ->
-                    {_SeqId, State2} = publish(Msg, true, false, State1),
+            fun (#msg_status { msg = Msg, 
+                               msg_properties = MsgProperties }, State1) ->
+                    {_SeqId, State2} = 
+                        publish(Msg, MsgPropsFun(MsgProperties), true, 
+                                false, State1),
                     State2;
-                ({IsPersistent, Guid}, State1) ->
+                ({IsPersistent, Guid, MsgProperties}, State1) ->
                     #vqstate { msg_store_clients = MSCState } = State1,
                     {{ok, Msg = #basic_message{}}, MSCState1} =
                         read_from_msg_store(MSCState, IsPersistent, Guid),
                     State2 = State1 #vqstate { msg_store_clients = MSCState1 },
-                    {_SeqId, State3} = publish(Msg, true, true, State2),
+                    {_SeqId, State3} = publish(Msg, MsgPropsFun(MsgProperties),
+                                               true, true, State2),
                     State3
             end,
             AckTags, State))).
@@ -783,10 +840,12 @@ one_if(false) -> 0.
 cons_if(true,   E, L) -> [E | L];
 cons_if(false, _E, L) -> L.
 
-msg_status(IsPersistent, SeqId, Msg = #basic_message { guid = Guid }) ->
+msg_status(IsPersistent, SeqId, Msg = #basic_message { guid = Guid }, 
+           MsgProperties) ->
     #msg_status { seq_id = SeqId, guid = Guid, msg = Msg,
                   is_persistent = IsPersistent, is_delivered = false,
-                  msg_on_disk = false, index_on_disk = false }.
+                  msg_on_disk = false, index_on_disk = false,
+                  msg_properties = MsgProperties }.
 
 find_msg_store(true)  -> ?PERSISTENT_MSG_STORE;
 find_msg_store(false) -> ?TRANSIENT_MSG_STORE.
@@ -821,24 +880,27 @@ store_tx(Txn, Tx) -> put({txn, Txn}, Tx).
 erase_tx(Txn) -> erase({txn, Txn}).
 
 persistent_guids(Pubs) ->
-    [Guid || #basic_message { guid = Guid, is_persistent = true } <- Pubs].
+    [Guid || 
+        {#basic_message { guid = Guid, is_persistent = true }, 
+          _MsgProps} <- Pubs].
 
 betas_from_index_entries(List, TransientThreshold, IndexState) ->
     {Filtered, Delivers, Acks} =
         lists:foldr(
-          fun ({Guid, SeqId, IsPersistent, IsDelivered},
+          fun ({Guid, SeqId, MsgProperties, IsPersistent, IsDelivered},
                {Filtered1, Delivers1, Acks1}) ->
                   case SeqId < TransientThreshold andalso not IsPersistent of
                       true  -> {Filtered1,
                                 cons_if(not IsDelivered, SeqId, Delivers1),
                                 [SeqId | Acks1]};
-                      false -> {[m(#msg_status { msg           = undefined,
-                                                 guid          = Guid,
-                                                 seq_id        = SeqId,
-                                                 is_persistent = IsPersistent,
-                                                 is_delivered  = IsDelivered,
-                                                 msg_on_disk   = true,
-                                                 index_on_disk = true
+                      false -> {[m(#msg_status { msg            = undefined,
+                                                 guid           = Guid,
+                                                 seq_id         = SeqId,
+                                                 is_persistent  = IsPersistent,
+                                                 is_delivered   = IsDelivered,
+                                                 msg_on_disk    = true,
+                                                 index_on_disk  = true,
+                                                 msg_properties = MsgProperties
                                                }) | Filtered1],
                                 Delivers1,
                                 Acks1}
@@ -885,11 +947,12 @@ update_rate(Now, Then, Count, {OThen, OCount}) ->
 %% Internal major helpers for Public API
 %%----------------------------------------------------------------------------
 
-msg_store_callback(PersistentGuids, Pubs, AckTags, Fun) ->
+msg_store_callback(PersistentGuids, Pubs, AckTags, Fun, MsgPropsFun) ->
     Self = self(),
     F = fun () -> rabbit_amqqueue:maybe_run_queue_via_backing_queue(
                     Self, fun (StateN) -> tx_commit_post_msg_store(
-                                            true, Pubs, AckTags, Fun, StateN)
+                                            true, Pubs, AckTags, 
+                                            Fun, MsgPropsFun, StateN)
                           end)
         end,
     fun () -> spawn(fun () -> ok = rabbit_misc:with_exit_handler(
@@ -900,7 +963,7 @@ msg_store_callback(PersistentGuids, Pubs, AckTags, Fun) ->
                     end)
     end.
 
-tx_commit_post_msg_store(HasPersistentPubs, Pubs, AckTags, Fun,
+tx_commit_post_msg_store(HasPersistentPubs, Pubs, AckTags, Fun, MsgPropsFun,
                          State = #vqstate {
                            on_sync     = OnSync = #sync {
                                            acks_persistent = SPAcks,
@@ -913,22 +976,27 @@ tx_commit_post_msg_store(HasPersistentPubs, Pubs, AckTags, Fun,
         case IsDurable of
             true  -> [AckTag || AckTag <- AckTags,
                                 case dict:fetch(AckTag, PA) of
-                                    #msg_status {}        -> false;
-                                    {IsPersistent, _Guid} -> IsPersistent
+                                    #msg_status {}     -> false;
+                                    {IsPersistent, 
+                                     _Guid, _MsgProps} -> IsPersistent
                                 end];
             false -> []
         end,
+    PubsOrdered = lists:foldl(
+                    fun ({Msg, MsgProps}, Acc) ->
+                            [{Msg, MsgPropsFun(MsgProps)} | Acc]
+                    end, [], Pubs),
     case IsDurable andalso (HasPersistentPubs orelse PersistentAcks =/= []) of
         true  -> State #vqstate { on_sync = #sync {
                                     acks_persistent = [PersistentAcks | SPAcks],
                                     acks_all        = [AckTags | SAcks],
-                                    pubs            = [Pubs | SPubs],
+                                    pubs            = [PubsOrdered | SPubs],
                                     funs            = [Fun | SFuns] }};
         false -> State1 = tx_commit_index(
                             State #vqstate { on_sync = #sync {
                                                acks_persistent = [],
                                                acks_all        = [AckTags],
-                                               pubs            = [Pubs],
+                                               pubs            = [PubsOrdered],
                                                funs            = [Fun] } }),
                  State1 #vqstate { on_sync = OnSync }
     end.
@@ -946,10 +1014,12 @@ tx_commit_index(State = #vqstate { on_sync = #sync {
     Pubs  = lists:append(lists:reverse(SPubs)),
     {SeqIds, State1 = #vqstate { index_state = IndexState }} =
         lists:foldl(
-          fun (Msg = #basic_message { is_persistent = IsPersistent },
+          fun ({Msg = #basic_message { is_persistent = IsPersistent }, 
+                MsgProperties},
                {SeqIdsAcc, State2}) ->
                   IsPersistent1 = IsDurable andalso IsPersistent,
-                  {SeqId, State3} = publish(Msg, false, IsPersistent1, State2),
+                  {SeqId, State3} = 
+                      publish(Msg, MsgProperties, false, IsPersistent1, State2),
                   {cons_if(IsPersistent1, SeqId, SeqIdsAcc), State3}
           end, {PAcks, ack(Acks, State)}, Pubs),
     IndexState1 = rabbit_queue_index:sync(SeqIds, IndexState),
@@ -996,7 +1066,7 @@ remove_queue_entries1(
 %%----------------------------------------------------------------------------
 
 publish(Msg = #basic_message { is_persistent = IsPersistent },
-        IsDelivered, MsgOnDisk,
+        MsgProperties, IsDelivered, MsgOnDisk,
         State = #vqstate { q1 = Q1, q3 = Q3, q4 = Q4,
                            next_seq_id      = SeqId,
                            len              = Len,
@@ -1005,8 +1075,9 @@ publish(Msg = #basic_message { is_persistent = IsPersistent },
                            durable          = IsDurable,
                            ram_msg_count    = RamMsgCount }) ->
     IsPersistent1 = IsDurable andalso IsPersistent,
-    MsgStatus = (msg_status(IsPersistent1, SeqId, Msg))
-        #msg_status { is_delivered = IsDelivered, msg_on_disk = MsgOnDisk },
+    MsgStatus = (msg_status(IsPersistent1, SeqId, Msg, MsgProperties))
+        #msg_status { is_delivered = IsDelivered, 
+                      msg_on_disk = MsgOnDisk},
     {MsgStatus1, State1} = maybe_write_to_disk(false, false, MsgStatus, State),
     State2 = case bpqueue:is_empty(Q3) of
                  false -> State1 #vqstate { q1 = queue:in(m(MsgStatus1), Q1) };
@@ -1045,12 +1116,18 @@ maybe_write_index_to_disk(_Force, MsgStatus = #msg_status {
     true = MsgStatus #msg_status.msg_on_disk, %% ASSERTION
     {MsgStatus, IndexState};
 maybe_write_index_to_disk(Force, MsgStatus = #msg_status {
-                                   guid = Guid, seq_id = SeqId,
+                                   guid = Guid,
+                                   seq_id = SeqId,
                                    is_persistent = IsPersistent,
-                                   is_delivered = IsDelivered }, IndexState)
+                                   is_delivered = IsDelivered,
+                                   msg_properties = MsgProperties},
+                          IndexState)
   when Force orelse IsPersistent ->
     true = MsgStatus #msg_status.msg_on_disk, %% ASSERTION
-    IndexState1 = rabbit_queue_index:publish(Guid, SeqId, IsPersistent,
+    IndexState1 = rabbit_queue_index:publish(Guid, 
+                                             SeqId, 
+                                             MsgProperties,
+                                             IsPersistent,
                                              IndexState),
     {MsgStatus #msg_status { index_on_disk = true },
      maybe_write_delivered(IsDelivered, SeqId, IndexState1)};
@@ -1073,9 +1150,11 @@ maybe_write_to_disk(ForceMsg, ForceIndex, MsgStatus,
 
 record_pending_ack(#msg_status { guid = Guid, seq_id = SeqId,
                                  is_persistent = IsPersistent,
-                                 msg_on_disk = MsgOnDisk } = MsgStatus, PA) ->
+                                 msg_on_disk = MsgOnDisk, 
+                                 msg_properties = MsgProperties } = MsgStatus, 
+                   PA) ->
     AckEntry = case MsgOnDisk of
-                   true  -> {IsPersistent, Guid};
+                   true  -> {IsPersistent, Guid, MsgProperties};
                    false -> MsgStatus
                end,
     dict:store(SeqId, AckEntry, PA).
@@ -1128,7 +1207,9 @@ accumulate_ack(_SeqId, #msg_status { is_persistent = false, %% ASSERTIONS
                                      msg_on_disk   = false,
                                      index_on_disk = false }, Acc) ->
     Acc;
-accumulate_ack(SeqId, {IsPersistent, Guid}, {SeqIdsAcc, Dict}) ->
+accumulate_ack(SeqId, 
+               {IsPersistent, Guid, _MsgProperties}, 
+               {SeqIdsAcc, Dict}) ->
     {cons_if(IsPersistent, SeqId, SeqIdsAcc),
      rabbit_misc:orddict_cons(find_msg_store(IsPersistent), Guid, Dict)}.
 
@@ -1224,40 +1305,31 @@ chunk_size(Current, Permitted)
 chunk_size(Current, Permitted) ->
     lists:min([Current - Permitted, ?IO_BATCH_SIZE]).
 
-fetch_from_q3_to_q4(State = #vqstate {
+fetch_from_q3(State = #vqstate {
                       q1                = Q1,
                       q2                = Q2,
                       delta             = #delta { count = DeltaCount },
                       q3                = Q3,
-                      q4                = Q4,
-                      ram_msg_count     = RamMsgCount,
-                      ram_index_count   = RamIndexCount,
-                      msg_store_clients = MSCState }) ->
+                      q4                = Q4 }) ->
     case bpqueue:out(Q3) of
         {empty, _Q3} ->
             {empty, State};
-        {{value, IndexOnDisk, MsgStatus = #msg_status {
-                                msg = undefined, guid = Guid,
-                                is_persistent = IsPersistent }}, Q3a} ->
-            {{ok, Msg = #basic_message {}}, MSCState1} =
-                read_from_msg_store(MSCState, IsPersistent, Guid),
-            Q4a = queue:in(m(MsgStatus #msg_status { msg = Msg }), Q4),
-            RamIndexCount1 = RamIndexCount - one_if(not IndexOnDisk),
-            true = RamIndexCount1 >= 0, %% ASSERTION
-            State1 = State #vqstate { q3                = Q3a,
-                                      q4                = Q4a,
-                                      ram_msg_count     = RamMsgCount + 1,
-                                      ram_index_count   = RamIndexCount1,
-                                      msg_store_clients = MSCState1 },
+        {{value, _IndexOnDisk, MsgStatus}, Q3a} ->
+
+            State1 = State #vqstate { q3 = Q3a},
+
             State2 =
                 case {bpqueue:is_empty(Q3a), 0 == DeltaCount} of
                     {true, true} ->
                         %% q3 is now empty, it wasn't before; delta is
-                        %% still empty. So q2 must be empty, and q1
-                        %% can now be joined onto q4
+                        %% still empty. So q2 must be empty, and we
+                        %% know q4 is empty otherwise we wouldn't be
+                        %% loading from q3. As such, we can just set
+                        %% q4 to Q1.
                         true = bpqueue:is_empty(Q2), %% ASSERTION
+                        true = queue:is_empty(Q4), %% ASSERTION
                         State1 #vqstate { q1 = queue:new(),
-                                          q4 = queue:join(Q4a, Q1) };
+                                          q4 = Q1 };
                     {true, false} ->
                         maybe_deltas_to_betas(State1);
                     {false, _} ->
@@ -1266,7 +1338,7 @@ fetch_from_q3_to_q4(State = #vqstate {
                         %% delta and q3 are maintained
                         State1
                 end,
-            {loaded, State2}
+            {loaded, {MsgStatus, State2}}
     end.
 
 maybe_deltas_to_betas(State = #vqstate { delta = ?BLANK_DELTA_PATTERN(X) }) ->