fix(gateway): truncate reads on an expired outbox entry (#2934)

fluxer_gateway/src/push/push_notification_format.erl

@@ -17,6 +17,7 @@
 
 -define(MAX_MENTIONS_FOR_PUSH, 50).
 -define(MAX_PREVIEW_BYTES, 100).
+-define(MAX_IMAGE_URL_BYTES, 1024).
 -define(CHANNEL_TYPE_GUILD_TEXT, 0).
 -define(CHANNEL_TYPE_GUILD_VOICE, 2).
 -define(CHANNEL_TYPE_GUILD_CATEGORY, 4).
@@ -487,11 +488,23 @@ format_name_list(Names) ->
 
 -spec extract_image_url(map()) -> binary() | undefined.
 extract_image_url(MessageData) ->
+    bounded_image_url(resolve_image_url(MessageData)).
+
+-spec resolve_image_url(map()) -> binary() | undefined.
+resolve_image_url(MessageData) ->
     case extract_attachment_image_url(maps:get(<<"attachments">>, MessageData, [])) of
         undefined -> extract_embed_image_url(maps:get(<<"embeds">>, MessageData, []));
         ImageUrl -> ImageUrl
     end.
 
+-spec bounded_image_url(binary() | undefined) -> binary() | undefined.
+bounded_image_url(ImageUrl) when
+    is_binary(ImageUrl), byte_size(ImageUrl) =< ?MAX_IMAGE_URL_BYTES
+->
+    ImageUrl;
+bounded_image_url(_ImageUrl) ->
+    undefined.
+
 -spec extract_attachment_image_url(term()) -> binary() | undefined.
 extract_attachment_image_url([Attachment | Rest]) when is_map(Attachment) ->
     case check_media_safe_url(Attachment) of
@@ -632,6 +645,30 @@ build_url_dm_test() ->
 build_url_guild_test() ->
     ?assertEqual(<<"/channels/123/456/789">>, build_url(123, 456, 789)).
 
+extract_image_url_keeps_a_url_within_the_size_bound_test() ->
+    Url = image_url_of_size(?MAX_IMAGE_URL_BYTES),
+    ?assertEqual(Url, extract_image_url(message_with_embed_image(Url))).
+
+extract_image_url_rejects_an_oversized_url_test() ->
+    Url = image_url_of_size(?MAX_IMAGE_URL_BYTES + 1),
+    ?assertEqual(undefined, extract_image_url(message_with_embed_image(Url))).
+
+image_url_of_size(Size) ->
+    Prefix = <<"https://media.example/">>,
+    <<Prefix/binary, (binary:copy(<<"a">>, Size - byte_size(Prefix)))/binary>>.
+
+message_with_embed_image(Url) ->
+    #{
+        <<"embeds">> => [
+            #{
+                <<"image">> => #{
+                    <<"content_type">> => <<"image/png">>,
+                    <<"proxy_url">> => Url
+                }
+            }
+        ]
+    }.
+
 extract_image_url_rejects_malformed_flags_test() ->
     MessageData = #{
         <<"attachments">> => [

fluxer_gateway/src/push/push_outbox.erl

@@ -437,10 +437,17 @@ handle_result({error, Reason}, Entry, State) ->
         attempts => maps:get(attempts, Entry)
     }),
     case is_expired(Entry, State) of
-        true -> fall_back(Entry, State);
+        true -> fall_back_prepared(Entry, State);
         false -> retry_settled(settle_if_stale(Entry, State))
     end.
 
+-spec fall_back_prepared(entry(), state()) -> state().
+fall_back_prepared(Entry, State) ->
+    case prepare(Entry, State) of
+        {skip, State1} -> State1;
+        {send, Prepared, State1} -> fall_back(Prepared, State1)
+    end.
+
 -spec retry_settled(settled()) -> state().
 retry_settled({keep, Entry, State}) ->
     schedule_retry(Entry, State);
@@ -661,3 +668,80 @@ app_pos_integer(Key, Default) ->
         Value when is_integer(Value), Value > 0 -> Value;
         _ -> Default
     end.
+
+-ifdef(TEST).
+-include_lib("eunit/include/eunit.hrl").
+
+an_expired_entry_is_handed_back_without_the_users_who_read_it_test() ->
+    Self = self(),
+    State = test_state(#{{7, 20} => {30, now_ms()}}),
+    Entry = test_entry([7, 8], fun(UserIds) -> Self ! {handed_back, UserIds} end),
+    Result = handle_result({error, timeout}, Entry, State),
+    ?assertEqual(1, count(fallbacks, Result)),
+    ?assertEqual(1, count(truncations, Result)),
+    receive
+        {handed_back, UserIds} -> ?assertEqual([8], UserIds)
+    after 2000 -> erlang:error(no_fallback_ran)
+    end.
+
+an_expired_entry_every_recipient_read_is_not_handed_back_test() ->
+    Self = self(),
+    State = test_state(#{{7, 20} => {30, now_ms()}, {8, 20} => {31, now_ms()}}),
+    Entry = test_entry([7, 8], fun(UserIds) -> Self ! {handed_back, UserIds} end),
+    Result = handle_result({error, timeout}, Entry, State),
+    ?assertEqual(0, count(fallbacks, Result)),
+    ?assertEqual(2, count(truncations, Result)),
+    receive
+        {handed_back, _} -> erlang:error(fallback_ran_for_read_users)
+    after 200 -> ok
+    end.
+
+an_expired_clear_is_handed_back_unchanged_test() ->
+    Self = self(),
+    State = test_state(#{{7, 20} => {30, now_ms()}, {8, 20} => {30, now_ms()}}),
+    Entry = (test_entry([7, 8], fun(UserIds) -> Self ! {handed_back, UserIds} end))#{
+        kind := clear
+    },
+    Result = handle_result({error, timeout}, Entry, State),
+    ?assertEqual(1, count(fallbacks, Result)),
+    receive
+        {handed_back, UserIds} -> ?assertEqual([7, 8], UserIds)
+    after 2000 -> erlang:error(no_fallback_ran)
+    end.
+
+test_state(Reads) ->
+    #{
+        jobs => gb_trees:empty(),
+        ready => queue:new(),
+        inflight => #{},
+        fallback_backlog => queue:new(),
+        fallback_runners => #{},
+        next_seq => 1,
+        reads => Reads,
+        active => #{},
+        counters => #{},
+        max_queue => ?DEFAULT_MAX_QUEUE,
+        max_inflight => ?DEFAULT_MAX_INFLIGHT,
+        max_fallback_runners => ?DEFAULT_MAX_FALLBACK_RUNNERS,
+        request_timeout_ms => ?DEFAULT_REQUEST_TIMEOUT_MS,
+        max_age_ms => ?DEFAULT_MAX_AGE_MS,
+        retry_base_ms => ?DEFAULT_RETRY_BASE_MS
+    }.
+
+test_entry(UserIds, Fallback) ->
+    #{
+        kind => message,
+        subject => <<"rpc.push.message">>,
+        job => #{<<"user_ids">> => [integer_to_binary(UserId) || UserId <- UserIds]},
+        body => <<"{}">>,
+        user_ids => UserIds,
+        channel_id => 20,
+        message_id => 30,
+        fallback => Fallback,
+        seq => 0,
+        enqueued_at => now_ms() - ?DEFAULT_MAX_AGE_MS,
+        attempts => 3,
+        config_version => undefined
+    }.
+
+-endif.