Merge pull request #203 from rabbitmq/last-committed-offset

kjnilsson · web-flow · commit 861df592a473 · 2026-01-26T14:58:57.000Z
Add last committed offset to stream stats
diff --git a/src/osiris.erl b/src/osiris.erl
@@ -40,6 +40,7 @@
       reference => term(),
       event_formatter => {module(), atom(), list()},
       retention => [osiris:retention_spec()],
+      features => features(),
       atom() => term()}.
 
 -type mfarg() :: {module(), atom(), list()}.
@@ -77,6 +78,7 @@
 -type data() :: iodata() |
                 batch() |
                 {filter_value(), iodata() | batch()}.
+-type features() :: #{committed_offset_calculate => boolean()}.
 
 %% returned when reading
 -type entry() :: binary() | batch().
@@ -307,13 +309,16 @@ configure_logger(Module) ->
     persistent_term:put('$osiris_logger', Module).
 
 -spec get_stats(pid()) -> #{committed_chunk_id => integer(),
-                            first_chunk_id => integer()}.
+                            first_chunk_id => integer(),
+                            last_chunk_id => integer(),
+                            committed_offset => integer()}.
 get_stats(Pid)
   when node(Pid) =:= node() ->
     #{shared := Shared} = osiris_util:get_reader_context(Pid),
     #{committed_chunk_id => osiris_log_shared:committed_chunk_id(Shared),
       first_chunk_id => osiris_log_shared:first_chunk_id(Shared),
-      last_chunk_id => osiris_log_shared:last_chunk_id(Shared)};
+      last_chunk_id => osiris_log_shared:last_chunk_id(Shared),
+      committed_offset => osiris_log_shared:committed_offset(Shared)};
 get_stats(Pid) when is_pid(Pid) ->
     erpc:call(node(Pid), ?MODULE, ?FUNCTION_NAME, [Pid]).
 
diff --git a/src/osiris.hrl b/src/osiris.hrl
@@ -30,6 +30,11 @@
                                                             domain => [osiris]}),
        ok).
 
+%% tail info pattern matching
+-define(TAIL_INFO(ChunkId, Ts), {_, {_, ChunkId, Ts}}).
+-define(TAIL_INFO(ChunkId), ?TAIL_INFO(ChunkId, _)).
+-define(TAIL_INFO_NEXT(NextOffset, ChunkId), {NextOffset, {_, ChunkId, _}}).
+
 -define(IS_STRING(S), is_list(S) orelse is_binary(S)).
 
 -define(C_NUM_LOG_FIELDS, 5).
diff --git a/src/osiris_counters.erl b/src/osiris_counters.erl
@@ -12,6 +12,7 @@
          fetch/1,
          overview/0,
          overview/1,
+         counters/2,
          delete/1
         ]).
 
@@ -45,3 +46,7 @@ overview() ->
 -spec overview(name()) -> #{atom() => non_neg_integer()} | undefined.
 overview(Name) ->
     seshat:counters(osiris, Name).
+
+-spec counters(name(), [atom()]) -> #{atom() => non_neg_integer()} | undefined.
+counters(Name, Fields) ->
+    seshat:counters(osiris, Name, Fields).
diff --git a/src/osiris_log.erl b/src/osiris_log.erl
@@ -36,9 +36,10 @@
          read_chunk/1,
          read_chunk_parsed/1,
          read_chunk_parsed/2,
-         committed_offset/1,
          committed_chunk_id/1,
+         committed_offset/1,
          set_committed_chunk_id/2,
+         set_committed_offset/2,
          last_chunk_id/1,
          get_current_epoch/1,
          get_directory/1,
@@ -1353,10 +1354,6 @@ last_user_chunk_id_in_index(NextPos, IdxFd) ->
             Error
     end.
 
--spec committed_offset(state()) -> integer().
-committed_offset(State) ->
-    committed_chunk_id(State).
-
 -spec committed_chunk_id(state()) -> integer().
 committed_chunk_id(#?MODULE{cfg = #cfg{shared = Ref}}) ->
     osiris_log_shared:committed_chunk_id(Ref).
@@ -1367,6 +1364,16 @@ set_committed_chunk_id(#?MODULE{mode = #write{},
   when is_integer(ChunkId) ->
     osiris_log_shared:set_committed_chunk_id(Ref, ChunkId).
 
+-spec set_committed_offset(state(), offset()) -> ok.
+set_committed_offset(#?MODULE{mode = #write{},
+                              cfg = #cfg{shared = Ref}}, Offset)
+  when is_integer(Offset) ->
+    osiris_log_shared:set_committed_offset(Ref, Offset).
+
+-spec committed_offset(state()) -> integer().
+committed_offset(#?MODULE{cfg = #cfg{shared = Ref}}) ->
+    osiris_log_shared:committed_offset(Ref).
+
 -spec last_chunk_id(state()) -> integer().
 last_chunk_id(#?MODULE{cfg = #cfg{shared = Ref}}) ->
     osiris_log_shared:last_chunk_id(Ref).
diff --git a/src/osiris_log_shared.erl b/src/osiris_log_shared.erl
@@ -1,34 +1,39 @@
 -module(osiris_log_shared).
 
--define(COMMITTED_IDX, 1).
+-define(COMMITTED_CHUNK_ID, 1).
 -define(FIRST_IDX, 2).
 -define(LAST_IDX, 3).
+-define(COMMITTED_OFFSET, 4).
 
 -export([
          new/0,
          committed_chunk_id/1,
          first_chunk_id/1,
          last_chunk_id/1,
+         committed_offset/1,
          set_committed_chunk_id/2,
          set_first_chunk_id/2,
-         set_last_chunk_id/2
+         set_last_chunk_id/2,
+         set_committed_offset/2
         ]).
 
 -type chunk_id() :: -1 | non_neg_integer().
+-type offset() :: -1 | non_neg_integer().
 
 -spec new() -> atomics:atomics_ref().
 new() ->
     %% Oh why, oh why did we think the first chunk id in
     %% a stream should have offset 0?
-    Ref = atomics:new(3, [{signed, true}]),
-    atomics:put(Ref, ?COMMITTED_IDX, -1),
+    Ref = atomics:new(4, [{signed, true}]),
+    atomics:put(Ref, ?COMMITTED_CHUNK_ID, -1),
     atomics:put(Ref, ?FIRST_IDX, -1),
     atomics:put(Ref, ?LAST_IDX, -1),
+    atomics:put(Ref, ?COMMITTED_OFFSET, -1),
     Ref.
 
 -spec committed_chunk_id(atomics:atomics_ref()) -> chunk_id().
 committed_chunk_id(Ref) ->
-    atomics:get(Ref, ?COMMITTED_IDX).
+    atomics:get(Ref, ?COMMITTED_CHUNK_ID).
 
 -spec first_chunk_id(atomics:atomics_ref()) -> chunk_id().
 first_chunk_id(Ref) ->
@@ -38,9 +43,13 @@ first_chunk_id(Ref) ->
 last_chunk_id(Ref) ->
     atomics:get(Ref, ?LAST_IDX).
 
+-spec committed_offset(atomics:atomics_ref()) -> chunk_id().
+committed_offset(Ref) ->
+    atomics:get(Ref, ?COMMITTED_OFFSET).
+
 -spec set_committed_chunk_id(atomics:atomics_ref(), chunk_id()) -> ok.
 set_committed_chunk_id(Ref, Value) when is_integer(Value) ->
-    atomics:put(Ref, ?COMMITTED_IDX, Value).
+    atomics:put(Ref, ?COMMITTED_CHUNK_ID, Value).
 
 -spec set_first_chunk_id(atomics:atomics_ref(), chunk_id()) -> ok.
 set_first_chunk_id(Ref, Value) when is_integer(Value) ->
@@ -50,6 +59,9 @@ set_first_chunk_id(Ref, Value) when is_integer(Value) ->
 set_last_chunk_id(Ref, Value) when is_integer(Value) ->
     atomics:put(Ref, ?LAST_IDX, Value).
 
+-spec set_committed_offset(atomics:atomics_ref(), offset()) -> ok.
+set_committed_offset(Ref, Value) when is_integer(Value) ->
+    atomics:put(Ref, ?COMMITTED_OFFSET, Value).
 
 
 -ifdef(TEST).
@@ -60,12 +72,15 @@ basics_test() ->
     ?assertEqual(-1, committed_chunk_id(R)),
     ?assertEqual(-1, first_chunk_id(R)),
     ?assertEqual(-1, last_chunk_id(R)),
+    ?assertEqual(-1, committed_offset(R)),
     ok = set_committed_chunk_id(R, 2),
+    ok = set_committed_offset(R, 3),
     ok = set_first_chunk_id(R, 1),
-    ok = set_last_chunk_id(R, 3),
+    ok = set_last_chunk_id(R, 4),
     ?assertEqual(2, committed_chunk_id(R)),
+    ?assertEqual(3, committed_offset(R)),
     ?assertEqual(1, first_chunk_id(R)),
-    ?assertEqual(3, last_chunk_id(R)),
+    ?assertEqual(4, last_chunk_id(R)),
 
     ok.
 
diff --git a/src/osiris_replica.erl b/src/osiris_replica.erl
@@ -50,7 +50,8 @@
          reference :: term(),
          event_formatter :: undefined | mfa(),
          counter :: counters:counters_ref(),
-         token :: undefined | binary()}).
+         token :: undefined | binary(),
+         committed_offset_calculate :: boolean()}).
 
 -type parse_state() ::
     undefined |
@@ -75,12 +76,15 @@
 -define(C_PACKETS, ?C_NUM_LOG_FIELDS + 3).
 -define(C_READERS, ?C_NUM_LOG_FIELDS + 4).
 -define(C_EPOCH, ?C_NUM_LOG_FIELDS + 5).
+-define(C_COMMITTED_CHUNK_ID, ?C_NUM_LOG_FIELDS + 6).
 -define(ADD_COUNTER_FIELDS,
         [{committed_offset, ?C_COMMITTED_OFFSET, counter, "Last committed offset"},
          {forced_gcs, ?C_FORCED_GCS, counter, "Number of garbage collection runs"},
          {packets, ?C_PACKETS, counter, "Number of packets"},
          {readers, ?C_READERS, counter, "Number of readers"},
-         {epoch, ?C_EPOCH, counter, "Current epoch"}]).
+         {epoch, ?C_EPOCH, counter, "Current epoch"},
+         {committed_chunk_id, ?C_COMMITTED_CHUNK_ID, counter, "Last committed chunk ID"}
+        ]).
 -define(FIELDSPEC_KEY, osiris_replica_seshat_fields_spec).
 
 -define(DEFAULT_ONE_TIME_TOKEN_TIMEOUT, 30000).
@@ -194,10 +198,10 @@ handle_continue(#{name := Name0,
             case LastChunk of
                 empty ->
                     ok;
-                {_, LastChId, LastTs} ->
+                _ ->
                     %% need to ack last chunk back to leader so that it can
                     %% re-discover the committed offset
-                    osiris_writer:ack(LeaderPid, {LastChId, LastTs})
+                    osiris_writer:ack(LeaderPid, ack_msg(Config, TailInfo))
             end,
             ?INFO_(Name, "osiris replica starting in epoch ~b, next offset ~b, tail info ~w",
                    [Epoch, NextOffset, TailInfo]),
@@ -236,6 +240,10 @@ handle_continue(#{name := Name0,
             Acceptor = spawn_link(fun() -> accept(Name, Transport, LSock, Self) end),
             ?DEBUG_(Name, "starting replica reader on node '~w'", [Node]),
 
+            CmttedOfstCalculate = committed_offset_calculate(Config),
+            Features0 = maps:get(features, Config, #{}),
+            Features1 = Features0#{committed_offset_calculate =>
+                                   committed_offset_calculate(Config)},
             ReplicaReaderConf = #{hosts => IpsHosts,
                                   port => Port,
                                   transport => Transport,
@@ -244,7 +252,8 @@ handle_continue(#{name := Name0,
                                   leader_pid => LeaderPid,
                                   start_offset => TailInfo,
                                   reference => ExtRef,
-                                  connection_token => Token},
+                                  connection_token => Token,
+                                  features => Features1},
             case osiris_replica_reader:start(Node, ReplicaReaderConf) of
                 {ok, RRPid} ->
                     true = link(RRPid),
@@ -260,6 +269,7 @@ handle_continue(#{name := Name0,
                                       false ->
                                           infinity
                                   end,
+                    counters:put(CntRef, ?C_COMMITTED_CHUNK_ID, -1),
                     counters:put(CntRef, ?C_COMMITTED_OFFSET, -1),
                     counters:put(CntRef, ?C_EPOCH, Epoch),
                     Shared = osiris_log:get_shared(Log),
@@ -281,7 +291,8 @@ handle_continue(#{name := Name0,
                                        event_formatter = EvtFmt,
                                        counter = CntRef,
                                        token = Token,
-                                       transport = Transport},
+                                       transport = Transport,
+                                       committed_offset_calculate = CmttedOfstCalculate},
                               log = Log,
                               parse_state = undefined}};
                 {error, {connection_refused = R, _}} ->
@@ -424,22 +435,26 @@ handle_call(Unknown, _From,
 %%                                  {stop, Reason, State}
 %% @end
 %%--------------------------------------------------------------------
-handle_cast({committed_offset, CommittedChId},
+handle_cast({committed_offset, {CommittedChId, LastOffset}},
             #?MODULE{cfg = #cfg{counter = Cnt},
                      log = Log,
                      committed_chunk_id = LastCommittedChId} =
                 State) ->
     case CommittedChId > LastCommittedChId of
         true ->
             %% notify offset listeners
-            counters:put(Cnt, ?C_COMMITTED_OFFSET, CommittedChId),
+            counters:put(Cnt, ?C_COMMITTED_CHUNK_ID, CommittedChId),
+            counters:put(Cnt, ?C_COMMITTED_OFFSET, LastOffset),
             ok = osiris_log:set_committed_chunk_id(Log, CommittedChId),
+            ok = osiris_log:set_committed_offset(Log, LastOffset),
             {noreply,
              notify_offset_listeners(
                State#?MODULE{committed_chunk_id = CommittedChId})};
         false ->
             State
     end;
+handle_cast({committed_offset, CommittedChId}, State) ->
+    handle_cast({committed_offset, {CommittedChId, -1}}, State);
 handle_cast({register_offset_listener, Pid, EvtFormatter, Offset},
             #?MODULE{cfg = #cfg{reference = Ref,
                                 event_formatter = DefaultFmt},
@@ -572,7 +587,7 @@ handle_incoming_data(Socket, Bin,
                               #cfg{socket = Socket,
                                    leader_pid = LeaderPid,
                                    transport = Transport,
-                                   counter = Cnt},
+                                   counter = Cnt} = Cfg,
                               parse_state = ParseState0,
                               log = Log0} =
                      State0) ->
@@ -594,7 +609,8 @@ handle_incoming_data(Socket, Bin,
         undefined ->
             {noreply, State1};
         _ ->
-            ok = osiris_writer:ack(LeaderPid, OffsetTimestamp),
+            TailInfo = osiris_log:tail_info(Log),
+            ok = osiris_writer:ack(LeaderPid, ack_msg(Cfg, TailInfo)),
             State = notify_offset_listeners(State1),
             {noreply, State}
     end.
@@ -738,7 +754,7 @@ notify_offset_listeners(#?MODULE{cfg = #cfg{reference = Ref,
     State#?MODULE{offset_listeners = L}.
 
 max_readable_chunk_id(Log) ->
-    min(osiris_log:committed_offset(Log), osiris_log:last_chunk_id(Log)).
+    min(osiris_log:committed_chunk_id(Log), osiris_log:last_chunk_id(Log)).
 
 %% INTERNAL
 
@@ -796,3 +812,20 @@ listen(ssl, Port, Options) ->
 init_fields_spec() ->
     persistent_term:put(?FIELDSPEC_KEY,
                         ?ADD_COUNTER_FIELDS ++ osiris_log:counter_fields()).
+
+ack_msg(Cfg, TailInfo) ->
+    case committed_offset_calculate(Cfg) of
+        true ->
+            TailInfo;
+        false ->
+            ?TAIL_INFO(TailChkId, TailTs) = TailInfo,
+            {TailChkId, TailTs}
+    end.
+
+committed_offset_calculate(#cfg{committed_offset_calculate = On}) ->
+    On;
+committed_offset_calculate(#{features := #{committed_offset_calculate := On}})
+  when is_boolean(On) ->
+    On;
+committed_offset_calculate(_) ->
+    false.
diff --git a/src/osiris_replica_reader.erl b/src/osiris_replica_reader.erl
diff --git a/src/osiris_writer.erl b/src/osiris_writer.erl
diff --git a/test/osiris_SUITE.erl b/test/osiris_SUITE.erl
