diff --git a/integration_test/unreadmarker_test.dart b/integration_test/unreadmarker_test.dart
index d51cfcc585..35d8c2ef40 100644
--- a/integration_test/unreadmarker_test.dart
+++ b/integration_test/unreadmarker_test.dart
@@ -28,7 +28,8 @@ void main() {
     final messages = List.generate(messageCount,
       (i) => eg.streamMessage(flags: [MessageFlag.read]));
     connection.prepare(json:
-      eg.newestGetMessagesResult(foundOldest: true, messages: messages).toJson());
+      eg.nearUnreadGetMessagesResult(foundOldest: true, foundNewest: true,
+        messages: messages).toJson());
 
     await tester.pumpWidget(TestZulipApp(accountId: eg.selfAccount.id,
       child: const MessageListPage(initNarrow: CombinedFeedNarrow())));
diff --git a/lib/model/message_list.dart b/lib/model/message_list.dart
index 670785ac4e..e4633a78b3 100644
--- a/lib/model/message_list.dart
+++ b/lib/model/message_list.dart
@@ -14,7 +14,7 @@ import 'narrow.dart';
 import 'store.dart';
 
 /// The number of messages to fetch in each request.
-const kMessageListFetchBatchSize = 100; // TODO tune
+const kMessageListFetchBatchSize = 5; // TODO tune
 
 /// A message, or one of its siblings shown in the message list.
 ///
@@ -58,7 +58,7 @@ class MessageListLoadingItem extends MessageListItem {
   const MessageListLoadingItem(this.direction);
 }
 
-enum MessageListDirection { older }
+enum MessageListDirection { older, newer }
 
 /// Indicates we've reached the oldest message in the narrow.
 class MessageListHistoryStartItem extends MessageListItem {
@@ -85,9 +85,6 @@ mixin _MessageSequence {
   bool _fetched = false;
 
   /// Whether we know we have the oldest messages for this narrow.
-  ///
-  /// (Currently we always have the newest messages for the narrow,
-  /// once [fetched] is true, because we start from the newest.)
   bool get haveOldest => _haveOldest;
   bool _haveOldest = false;
 
@@ -118,6 +115,40 @@ mixin _MessageSequence {
 
   BackoffMachine? _fetchOlderCooldownBackoffMachine;
 
+  /// Whether we know we have the newest messages for this narrow.
+  bool get haveNewest => _haveNewest;
+  bool _haveNewest = false;
+
+  /// Whether we are currently fetching the next batch of newer messages.
+  ///
+  /// When this is true, [fetchNewer] is a no-op.
+  /// That method is called frequently by Flutter's scrolling logic,
+  /// and this field helps us avoid spamming the same request just to get
+  /// the same response each time.
+  ///
+  /// See also [fetchNewerCoolingDown].
+  bool get fetchingNewer => _fetchingNewer;
+  bool _fetchingNewer = false;
+
+  /// Whether [fetchNewer] had a request error recently.
+  ///
+  /// When this is true, [fetchNewer] is a no-op.
+  /// That method is called frequently by Flutter's scrolling logic,
+  /// and this field mitigates spamming the same request and getting
+  /// the same error each time.
+  ///
+  /// "Recently" is decided by a [BackoffMachine] that resets
+  /// when a [fetchNewer] request succeeds.
+  ///
+  /// See also [fetchingNewer].
+  bool get fetchNewerCoolingDown => _fetchNewerCoolingDown;
+  bool _fetchNewerCoolingDown = false;
+
+  BackoffMachine? _fetchNewerCooldownBackoffMachine;
+
+  int? get firstUnreadMessageId => _firstUnreadMessageId;
+  int? _firstUnreadMessageId;
+
   /// The parsed message contents, as a list parallel to [messages].
   ///
   /// The i'th element is the result of parsing the i'th element of [messages].
@@ -151,6 +182,7 @@ mixin _MessageSequence {
       case MessageListHistoryStartItem():        return -1;
       case MessageListLoadingItem():
         switch (item.direction) {
+          case MessageListDirection.newer:       return 1;
           case MessageListDirection.older:       return -1;
         }
       case MessageListRecipientHeaderItem(:var message):
@@ -269,6 +301,11 @@ mixin _MessageSequence {
     _fetchingOlder = false;
     _fetchOlderCoolingDown = false;
     _fetchOlderCooldownBackoffMachine = null;
+    _haveNewest = false;
+    _fetchingNewer = false;
+    _fetchNewerCoolingDown = false;
+    _fetchNewerCooldownBackoffMachine = null;
+    _firstUnreadMessageId = null;
     contents.clear();
     items.clear();
   }
@@ -317,7 +354,8 @@ mixin _MessageSequence {
   /// Update [items] to include markers at start and end as appropriate.
   void _updateEndMarkers() {
     assert(fetched);
-    assert(!(fetchingOlder && fetchOlderCoolingDown));
+    assert(!(fetchingOlder && fetchOlderCoolingDown)
+        || !(fetchingNewer && fetchNewerCoolingDown));
     final effectiveFetchingOlder = fetchingOlder || fetchOlderCoolingDown;
     assert(!(effectiveFetchingOlder && haveOldest));
     final startMarker = switch ((effectiveFetchingOlder, haveOldest)) {
@@ -336,6 +374,21 @@ mixin _MessageSequence {
       case (_,    true): items.removeFirst();
       case (_,    _   ): break;
     }
+
+    final effectiveFetchingNewer = fetchingNewer || fetchNewerCoolingDown;
+    final endMarker = switch (effectiveFetchingNewer) {
+      true => const MessageListLoadingItem(MessageListDirection.newer),
+      false => null,
+    };
+    final hasEndMarker = switch (items.lastOrNull) {
+      MessageListLoadingItem() => true,
+      _                        => false,
+    };
+    switch ((endMarker != null, hasEndMarker)) {
+      case (true,  false): items.add(endMarker!);
+      case (false, true ): items.removeLast();
+      case (_,     _    ): break;
+    }
   }
 
   /// Recompute [items] from scratch, based on [messages], [contents], and flags.
@@ -500,15 +553,17 @@ class MessageListView with ChangeNotifier, _MessageSequence {
   Future<void> fetchInitial() async {
     // TODO(#80): fetch from anchor firstUnread, instead of newest
     // TODO(#82): fetch from a given message ID as anchor
-    assert(!fetched && !haveOldest && !fetchingOlder && !fetchOlderCoolingDown);
+    assert(!fetched && !haveOldest && !fetchingOlder && !fetchOlderCoolingDown
+                    && !haveNewest && !fetchingNewer && !fetchNewerCoolingDown);
     assert(messages.isEmpty && contents.isEmpty);
     // TODO schedule all this in another isolate
     final generation = this.generation;
     final result = await getMessages(store.connection,
       narrow: narrow.apiEncode(),
-      anchor: AnchorCode.newest,
+      anchor: AnchorCode.firstUnread,
       numBefore: kMessageListFetchBatchSize,
-      numAfter: 0,
+      // Results will include the anchor message, so fetch one less.
+      numAfter: kMessageListFetchBatchSize - 1,
     );
     if (this.generation > generation) return;
     store.reconcileMessages(result.messages);
@@ -520,6 +575,8 @@ class MessageListView with ChangeNotifier, _MessageSequence {
     }
     _fetched = true;
     _haveOldest = result.foundOldest;
+    _haveNewest = result.foundNewest;
+    _firstUnreadMessageId = result.anchor;
     _updateEndMarkers();
     notifyListeners();
   }
@@ -541,7 +598,7 @@ class MessageListView with ChangeNotifier, _MessageSequence {
       try {
         result = await getMessages(store.connection,
           narrow: narrow.apiEncode(),
-          anchor: NumericAnchor(messages[0].id),
+          anchor: NumericAnchor(messages.first.id),
           includeAnchor: false,
           numBefore: kMessageListFetchBatchSize,
           numAfter: 0,
@@ -553,7 +610,7 @@ class MessageListView with ChangeNotifier, _MessageSequence {
       if (this.generation > generation) return;
 
       if (result.messages.isNotEmpty
-          && result.messages.last.id == messages[0].id) {
+          && result.messages.last.id == messages.first.id) {
         // TODO(server-6): includeAnchor should make this impossible
         result.messages.removeLast();
       }
@@ -589,6 +646,71 @@ class MessageListView with ChangeNotifier, _MessageSequence {
     }
   }
 
+  /// Fetch the next batch of newer messages, if applicable.
+  Future<void> fetchNewer() async {
+    if (haveNewest) return;
+    if (fetchingNewer) return;
+    if (fetchNewerCoolingDown) return;
+    assert(fetched);
+    assert(messages.isNotEmpty);
+    _fetchingNewer = true;
+    _updateEndMarkers();
+    notifyListeners();
+    final generation = this.generation;
+    bool hasFetchError = false;
+    try {
+      final GetMessagesResult result;
+      try {
+        result = await getMessages(store.connection,
+          narrow: narrow.apiEncode(),
+          anchor: NumericAnchor(messages.last.id),
+          includeAnchor: false,
+          numBefore: 0,
+          numAfter: kMessageListFetchBatchSize,
+        );
+      } catch (e) {
+        hasFetchError = true;
+        rethrow;
+      }
+      if (this.generation > generation) return;
+
+      if (result.messages.isNotEmpty
+          && result.messages.first.id == messages.last.id) {
+        // TODO(server-6): includeAnchor should make this impossible
+        result.messages.removeAt(0);
+      }
+
+      store.reconcileMessages(result.messages);
+      store.recentSenders.handleMessages(result.messages); // TODO(#824)
+
+      final fetchedMessages = _allMessagesVisible
+        ? result.messages // Avoid unnecessarily copying the list.
+        : result.messages.where(_messageVisible);
+
+      _insertAllMessages(messages.length, fetchedMessages);
+      _haveNewest = result.foundNewest;
+    } finally {
+      if (this.generation == generation) {
+        _fetchingNewer = false;
+        if (hasFetchError) {
+          assert(!fetchNewerCoolingDown);
+          _fetchNewerCoolingDown = true;
+          unawaited((_fetchNewerCooldownBackoffMachine ??= BackoffMachine())
+            .wait().then((_) {
+              if (this.generation != generation) return;
+              _fetchNewerCoolingDown = false;
+              _updateEndMarkers();
+              notifyListeners();
+            }));
+        } else {
+          _fetchNewerCooldownBackoffMachine = null;
+        }
+        _updateEndMarkers();
+        notifyListeners();
+      }
+    }
+  }
+
   void handleUserTopicEvent(UserTopicEvent event) {
     switch (_canAffectVisibility(event)) {
       case VisibilityEffect.none:
diff --git a/lib/widgets/message_list.dart b/lib/widgets/message_list.dart
index 8bfacf9a3d..2ebf03603a 100644
--- a/lib/widgets/message_list.dart
+++ b/lib/widgets/message_list.dart
@@ -509,6 +509,9 @@ class _MessageListState extends State<MessageList> with PerAccountStoreAwareStat
       //   still not yet updated to account for the newly-added messages.
       model?.fetchOlder();
     }
+    if (scrollMetrics.extentAfter < kFetchMessagesBufferPixels) {
+      model?.fetchNewer();
+    }
   }
 
   void _scrollChanged() {
@@ -566,10 +569,27 @@ class _MessageListState extends State<MessageList> with PerAccountStoreAwareStat
   }
 
   Widget _buildListView(BuildContext context) {
-    final length = model!.items.length;
     const centerSliverKey = ValueKey('center sliver');
 
-    Widget sliver = SliverStickyHeaderList(
+    // TODO(#311) If we have a bottom nav, it will pad the bottom
+    //   inset, and this shouldn't be necessary
+    final hasComposeBox = ComposeBox.hasComposeBox(widget.narrow);
+
+    // TODO make sticky headers work
+    final List<MessageListItem> olderItems, newerItems;
+
+    final firstUnreadItemIndex = model!.firstUnreadMessageId! != 10000000000000000
+        ? model!.findItemWithMessageId(model!.firstUnreadMessageId!)
+        : -1;
+    if (firstUnreadItemIndex == -1) {
+      olderItems = model!.items;
+      newerItems = const [];
+    } else {
+      olderItems = model!.items.slice(0, firstUnreadItemIndex);
+      newerItems = model!.items.slice(firstUnreadItemIndex);
+    }
+
+    Widget topSliver = SliverStickyHeaderList(
       headerPlacement: HeaderPlacement.scrollingStart,
       delegate: SliverChildBuilderDelegate(
         // To preserve state across rebuilds for individual [MessageItem]
@@ -591,26 +611,56 @@ class _MessageListState extends State<MessageList> with PerAccountStoreAwareStat
           final valueKey = key as ValueKey<int>;
           final index = model!.findItemWithMessageId(valueKey.value);
           if (index == -1) return null;
-          return length - 1 - (index - 3);
+          return olderItems.length - 1 - index;
         },
-        childCount: length + 3,
+        childCount: olderItems.length,
         (context, i) {
-          // To reinforce that the end of the feed has been reached:
-          //   https://chat.zulip.org/#narrow/stream/243-mobile-team/topic/flutter.3A.20Mark-as-read/near/1680603
-          if (i == 0) return const SizedBox(height: 36);
+          final data = olderItems[olderItems.length - 1 - i];
+          return _buildItem(data, i);
+        }));
+
+    Widget bottomSliver = SliverStickyHeaderList(
+      key: hasComposeBox ? centerSliverKey : null,
+      headerPlacement: HeaderPlacement.scrollingStart,
+      delegate: SliverChildBuilderDelegate(
+        // To preserve state across rebuilds for individual [MessageItem]
+        // widgets as the size of [MessageListView.items] changes we need
+        // to match old widgets by their key to their new position in
+        // the list.
+        //
+        // The keys are of type [ValueKey] with a value of [Message.id]
+        // and here we use a O(log n) binary search method. This could
+        // be improved but for now it only triggers for materialized
+        // widgets. As a simple test, flinging through Combined feed in
+        // CZO on a Pixel 5, this only runs about 10 times per rebuild
+        // and the timing for each call is <100 microseconds.
+        //
+        // Non-message items (e.g., start and end markers) that do not
+        // have state that needs to be preserved have not been given keys
+        // and will not trigger this callback.
+        findChildIndexCallback: (Key key) {
+          final valueKey = key as ValueKey<int>;
+          final index = model!.findItemWithMessageId(valueKey.value);
+          if (index == -1) return null;
+          return index;
+        },
+        childCount: newerItems.length + 3,
+        (context, i) {
+          if (i == newerItems.length) return TypingStatusWidget(narrow: widget.narrow);
 
-          if (i == 1) return MarkAsReadWidget(narrow: widget.narrow);
+          if (i == newerItems.length + 1) return MarkAsReadWidget(narrow: widget.narrow);
 
-          if (i == 2) return TypingStatusWidget(narrow: widget.narrow);
+          // To reinforce that the end of the feed has been reached:
+          //   https://chat.zulip.org/#narrow/stream/243-mobile-team/topic/flutter.3A.20Mark-as-read/near/1680603
+          if (i == newerItems.length + 2) return const SizedBox(height: 36);
 
-          final data = model!.items[length - 1 - (i - 3)];
+          final data = newerItems[i];
           return _buildItem(data, i);
         }));
 
-    if (!ComposeBox.hasComposeBox(widget.narrow)) {
-      // TODO(#311) If we have a bottom nav, it will pad the bottom inset,
-      //   and this can be removed; also remove mention in MessageList dartdoc
-      sliver = SliverSafeArea(sliver: sliver);
+    if (!hasComposeBox) {
+      topSliver = SliverSafeArea(sliver: topSliver);
+      bottomSliver = SliverSafeArea(key: centerSliverKey, sliver: bottomSliver);
     }
 
     return CustomScrollView(
@@ -626,17 +676,12 @@ class _MessageListState extends State<MessageList> with PerAccountStoreAwareStat
       },
 
       controller: scrollController,
-      semanticChildCount: length + 2,
-      anchor: 1.0,
+      semanticChildCount: model!.items.length + 2,
+      anchor: 0.75,
       center: centerSliverKey,
-
       slivers: [
-        sliver,
-
-        // This is a trivial placeholder that occupies no space.  Its purpose is
-        // to have the key that's passed to [ScrollView.center], and so to cause
-        // the above [SliverStickyHeaderList] to run from bottom to top.
-        const SliverToBoxAdapter(key: centerSliverKey),
+        topSliver,
+        bottomSliver,
       ]);
   }
 
diff --git a/test/example_data.dart b/test/example_data.dart
index 0e45321632..310339352c 100644
--- a/test/example_data.dart
+++ b/test/example_data.dart
@@ -468,20 +468,31 @@ DmMessage dmMessage({
   }) as Map<String, dynamic>);
 }
 
-/// A GetMessagesResult the server might return on an `anchor=newest` request.
-GetMessagesResult newestGetMessagesResult({
+/// A GetMessagesResult the server might return on an `anchor=first_unread` request.
+///
+/// The expected [messages] list must be non-empty.
+GetMessagesResult nearUnreadGetMessagesResult({
+  required bool foundNewest,
   required bool foundOldest,
+  bool foundAnchor = true,
   bool historyLimited = false,
   required List<Message> messages,
 }) {
+  if (messages.isEmpty) {
+    return GetMessagesResult(
+      anchor: 10000000000000000,
+      foundNewest: true,
+      foundOldest: true,
+      foundAnchor: false,
+      historyLimited: historyLimited,
+      messages: const [],
+    );
+  }
   return GetMessagesResult(
-    // These anchor, foundAnchor, and foundNewest values are what the server
-    // appears to always return when the request had `anchor=newest`.
-    anchor: 10000000000000000, // that's 16 zeros
-    foundAnchor: false,
-    foundNewest: true,
-
+    anchor: messages[messages.length ~/ 2].id,
+    foundNewest: foundNewest,
     foundOldest: foundOldest,
+    foundAnchor: foundAnchor,
     historyLimited: historyLimited,
     messages: messages,
   );
@@ -505,6 +516,24 @@ GetMessagesResult olderGetMessagesResult({
   );
 }
 
+/// A GetMessagesResult the server might return when we request newer messages.
+GetMessagesResult newerGetMessagesResult({
+  required int anchor,
+  bool foundAnchor = false, // the value if the server understood includeAnchor false
+  required bool foundNewest,
+  bool historyLimited = false,
+  required List<Message> messages,
+}) {
+  return GetMessagesResult(
+    anchor: anchor,
+    foundAnchor: foundAnchor,
+    foundNewest: foundNewest,
+    foundOldest: false, // empirically always this, even when anchor happens to be oldest
+    historyLimited: historyLimited,
+    messages: messages,
+  );
+}
+
 PollWidgetData pollWidgetData({
   required String question,
   required List<String> options,
diff --git a/test/model/message_list_test.dart b/test/model/message_list_test.dart
index 6a1d103c84..90d7e38c0f 100644
--- a/test/model/message_list_test.dart
+++ b/test/model/message_list_test.dart
@@ -23,8 +23,9 @@ import 'content_checks.dart';
 import 'recent_senders_test.dart' as recent_senders_test;
 import 'test_store.dart';
 
-const newestResult = eg.newestGetMessagesResult;
+const nearUnreadResult = eg.nearUnreadGetMessagesResult;
 const olderResult = eg.olderGetMessagesResult;
+const newerResult = eg.newerGetMessagesResult;
 
 void main() {
   // These variables are the common state operated on by each test.
@@ -68,10 +69,14 @@ void main() {
   /// The test case must have already called [prepare] to initialize the state.
   Future<void> prepareMessages({
     required bool foundOldest,
+    required bool foundNewest,
     required List<Message> messages,
   }) async {
     connection.prepare(json:
-      newestResult(foundOldest: foundOldest, messages: messages).toJson());
+      nearUnreadResult(
+        foundOldest: foundOldest,
+        foundNewest: foundNewest,
+        messages: messages).toJson());
     await model.fetchInitial();
     checkNotifiedOnce();
   }
@@ -96,11 +101,13 @@ void main() {
   }
 
   test('fetchInitial', () async {
+    // TODO replace with prepareWithInitialMessages
     const narrow = CombinedFeedNarrow();
     await prepare(narrow: narrow);
-    connection.prepare(json: newestResult(
+    connection.prepare(json: nearUnreadResult(
       foundOldest: false,
-      messages: List.generate(kMessageListFetchBatchSize,
+      foundNewest: false,
+      messages: List.generate(kMessageListFetchBatchSize * 2,
         (i) => eg.streamMessage()),
     ).toJson());
     final fetchFuture = model.fetchInitial();
@@ -110,20 +117,21 @@ void main() {
     await fetchFuture;
     checkNotifiedOnce();
     check(model)
-      ..messages.length.equals(kMessageListFetchBatchSize)
+      ..messages.length.equals(kMessageListFetchBatchSize * 2)
       ..haveOldest.isFalse();
     checkLastRequest(
       narrow: narrow.apiEncode(),
-      anchor: 'newest',
+      anchor: 'first_unread',
       numBefore: kMessageListFetchBatchSize,
-      numAfter: 0,
+      numAfter: kMessageListFetchBatchSize - 1,
     );
   });
 
   test('fetchInitial, short history', () async {
     await prepare();
-    connection.prepare(json: newestResult(
+    connection.prepare(json: nearUnreadResult(
       foundOldest: true,
+      foundNewest: true,
       messages: List.generate(30, (i) => eg.streamMessage()),
     ).toJson());
     await model.fetchInitial();
@@ -135,8 +143,9 @@ void main() {
 
   test('fetchInitial, no messages found', () async {
     await prepare();
-    connection.prepare(json: newestResult(
+    connection.prepare(json: nearUnreadResult(
       foundOldest: true,
+      foundNewest: true,
       messages: [],
     ).toJson());
     await model.fetchInitial();
@@ -156,8 +165,9 @@ void main() {
       // Not subscribed to the stream with id 10.
       eg.streamMessage(stream: eg.stream(streamId: 10)),
     ];
-    connection.prepare(json: newestResult(
+    connection.prepare(json: nearUnreadResult(
       foundOldest: false,
+      foundNewest: false,
       messages: messages,
     ).toJson());
     await model.fetchInitial();
@@ -169,7 +179,7 @@ void main() {
   test('fetchOlder', () async {
     const narrow = CombinedFeedNarrow();
     await prepare(narrow: narrow);
-    await prepareMessages(foundOldest: false,
+    await prepareMessages(foundOldest: false, foundNewest: false,
       messages: List.generate(100, (i) => eg.streamMessage(id: 1000 + i)));
 
     connection.prepare(json: olderResult(
@@ -197,7 +207,7 @@ void main() {
   test('fetchOlder nop when already fetching', () async {
     const narrow = CombinedFeedNarrow();
     await prepare(narrow: narrow);
-    await prepareMessages(foundOldest: false,
+    await prepareMessages(foundOldest: false, foundNewest: false,
       messages: List.generate(100, (i) => eg.streamMessage(id: 1000 + i)));
 
     connection.prepare(json: olderResult(
@@ -225,8 +235,8 @@ void main() {
 
   test('fetchOlder nop when already haveOldest true', () async {
     await prepare(narrow: const CombinedFeedNarrow());
-    await prepareMessages(foundOldest: true, messages:
-      List.generate(30, (i) => eg.streamMessage()));
+    await prepareMessages(foundOldest: true, foundNewest: false,
+      messages: List.generate(30, (i) => eg.streamMessage()));
     check(model)
       ..haveOldest.isTrue()
       ..messages.length.equals(30);
@@ -244,7 +254,7 @@ void main() {
     final olderMessages = List.generate(5, (i) => eg.streamMessage());
     final initialMessages = List.generate(5, (i) => eg.streamMessage());
     await prepare(narrow: const CombinedFeedNarrow());
-    await prepareMessages(foundOldest: false, messages: initialMessages);
+    await prepareMessages(foundOldest: false, foundNewest: false, messages: initialMessages);
     check(connection.takeRequests()).single;
 
     connection.prepare(httpStatus: 400, json: {
@@ -277,7 +287,7 @@ void main() {
   test('fetchOlder handles servers not understanding includeAnchor', () async {
     const narrow = CombinedFeedNarrow();
     await prepare(narrow: narrow);
-    await prepareMessages(foundOldest: false,
+    await prepareMessages(foundOldest: false, foundNewest: false,
       messages: List.generate(100, (i) => eg.streamMessage(id: 1000 + i)));
 
     // The old behavior is to include the anchor message regardless of includeAnchor.
@@ -297,7 +307,7 @@ void main() {
     const narrow = CombinedFeedNarrow();
     await prepare(narrow: narrow);
     final initialMessages = List.generate(10, (i) => eg.streamMessage(id: 100 + i));
-    await prepareMessages(foundOldest: false, messages: initialMessages);
+    await prepareMessages(foundOldest: false, foundNewest: false, messages: initialMessages);
 
     final oldMessages = List.generate(10, (i) => eg.streamMessage(id: 89 + i))
       // Not subscribed to the stream with id 10.
@@ -313,11 +323,163 @@ void main() {
       [...initialMessages, ...oldMessages]);
   });
 
+  test('fetchNewer', () async {
+    const narrow = CombinedFeedNarrow();
+    await prepare(narrow: narrow);
+    final initialMessages = List.generate(100, (i) => eg.streamMessage(id: 1000 + i));
+    await prepareMessages(foundOldest: false, foundNewest: false, messages: initialMessages);
+
+    connection.prepare(json: newerResult(
+      anchor: 1099,
+      foundNewest: false,
+      messages: List.generate(100, (i) => eg.streamMessage(id: 1100 + i))).toJson());
+    final fetchFuture = model.fetchNewer();
+    checkNotifiedOnce();
+    check(model).fetchingNewer.isTrue();
+
+    await fetchFuture;
+    checkNotifiedOnce();
+    check(model)
+      ..fetchingNewer.isFalse()
+      ..messages.length.equals(200);
+    checkLastRequest(
+      narrow: narrow.apiEncode(),
+      anchor: '1099',
+      includeAnchor: false,
+      numBefore: 0,
+      numAfter: kMessageListFetchBatchSize,
+    );
+  });
+
+  test('fetchNewer nop when already fetching', () async {
+    const narrow = CombinedFeedNarrow();
+    await prepare(narrow: narrow);
+    final initialMessages = List.generate(100, (i) => eg.streamMessage(id: 1000 + i));
+    await prepareMessages(foundOldest: false, foundNewest: false, messages: initialMessages);
+    check(model).messages.length.equals(100);
+
+    connection.prepare(json: newerResult(
+      anchor: 1099,
+      foundNewest: false,
+      messages: List.generate(100, (i) => eg.streamMessage(id: 1100 + i))).toJson());
+    final fetchFuture = model.fetchNewer();
+    checkNotifiedOnce();
+    check(model).fetchingNewer.isTrue();
+
+    // Don't prepare another response.
+    final fetchFuture2 = model.fetchNewer();
+    checkNotNotified();
+    check(model).fetchingNewer.isTrue();
+
+    await fetchFuture;
+    await fetchFuture2;
+    // We must not have made another request, because we didn't
+    // prepare another response and didn't get an exception.
+    checkNotifiedOnce();
+    check(model)
+      ..fetchingNewer.isFalse()
+      ..messages.length.equals(200);
+  });
+
+  test('fetchNewer nop when already haveNewest true', () async {
+    await prepare(narrow: const CombinedFeedNarrow());
+    await prepareMessages(
+      foundOldest: false,
+      foundNewest: true,
+      messages: List.generate(30, (i) => eg.streamMessage()));
+    check(model)
+      ..haveNewest.isTrue()
+      ..messages.length.equals(30);
+
+    await model.fetchNewer();
+    // We must not have made a request, because we didn't
+    // prepare a response and didn't get an exception.
+    checkNotNotified();
+    check(model)
+      ..haveNewest.isTrue()
+      ..messages.length.equals(30);
+  });
+
+  test('fetchNewer nop during backoff', () => awaitFakeAsync((async) async {
+    final initialMessages = List.generate(5, (i) => eg.streamMessage());
+    final newerMessages = List.generate(5, (i) => eg.streamMessage());
+    await prepare(narrow: const CombinedFeedNarrow());
+    await prepareMessages(foundOldest: false, foundNewest: false, messages: initialMessages);
+    check(connection.takeRequests()).single;
+
+    connection.prepare(httpStatus: 400, json: {
+      'result': 'error', 'code': 'BAD_REQUEST', 'msg': 'Bad request'});
+    await check(model.fetchNewer()).throws<ZulipApiException>();
+    async.elapse(Duration.zero);
+    checkNotified(count: 2);
+    check(model).fetchNewerCoolingDown.isTrue();
+    check(connection.takeRequests()).single;
+
+    await model.fetchNewer();
+    checkNotNotified();
+    check(model).fetchNewerCoolingDown.isTrue();
+    check(model).fetchingNewer.isFalse();
+    check(connection.lastRequest).isNull();
+
+    // Wait long enough that a first backoff is sure to finish.
+    async.elapse(const Duration(seconds: 1));
+    check(model).fetchNewerCoolingDown.isFalse();
+    checkNotifiedOnce();
+    check(connection.lastRequest).isNull();
+
+    connection.prepare(json: newerResult(
+      anchor: 1000, foundNewest: false, messages: newerMessages).toJson());
+    await model.fetchNewer();
+    checkNotified(count: 2);
+    check(connection.takeRequests()).single;
+  }));
+
+  test('fetchNewer handles servers not understanding includeAnchor', () async {
+    const narrow = CombinedFeedNarrow();
+    await prepare(narrow: narrow);
+    await prepareMessages(foundOldest: false, foundNewest: false,
+      messages: List.generate(100, (i) => eg.streamMessage(id: 1000 + i)));
+
+    // The old behavior is to include the anchor message regardless of includeAnchor.
+    connection.prepare(json: newerResult(
+      anchor: 1099, foundNewest: false, foundAnchor: true,
+      messages: List.generate(101, (i) => eg.streamMessage(id: 1099 + i)),
+    ).toJson());
+    await model.fetchNewer();
+    checkNotified(count: 2);
+    check(model)
+      ..fetchingNewer.isFalse()
+      ..messages.length.equals(200);
+  });
+
+  // TODO(#824): move this test
+  test('fetchNewer, recent senders track all the messages', () async {
+    const narrow = CombinedFeedNarrow();
+    await prepare(narrow: narrow);
+    final initialMessages = List.generate(10, (i) => eg.streamMessage(id: 100 + i));
+    await prepareMessages(foundOldest: false, foundNewest: false, messages: initialMessages);
+
+    final newMessages = [
+      // Not subscribed to the stream with id 10.
+      eg.streamMessage(id: 110, stream: eg.stream(streamId: 10)),
+      ...List.generate(10, (i) => eg.streamMessage(id: 111 + i)),
+    ];
+    connection.prepare(json: newerResult(
+      anchor: 109, foundNewest: false,
+      messages: newMessages,
+    ).toJson());
+    await model.fetchNewer();
+
+    check(model).messages.length.equals(20);
+    recent_senders_test.checkMatchesMessages(store.recentSenders,
+      [...initialMessages, ...newMessages]);
+  });
+
   test('MessageEvent', () async {
     final stream = eg.stream();
     await prepare(narrow: ChannelNarrow(stream.streamId));
-    await prepareMessages(foundOldest: true, messages:
-      List.generate(30, (i) => eg.streamMessage(stream: stream)));
+    await prepareMessages(foundOldest: true, foundNewest: false,
+     messages: List.generate(30, (i) => eg.streamMessage(stream: stream)));
 
     check(model).messages.length.equals(30);
     await store.handleEvent(MessageEvent(id: 0,
@@ -329,8 +491,8 @@ void main() {
   test('MessageEvent, not in narrow', () async {
     final stream = eg.stream();
     await prepare(narrow: ChannelNarrow(stream.streamId));
-    await prepareMessages(foundOldest: true, messages:
-      List.generate(30, (i) => eg.streamMessage(stream: stream)));
+    await prepareMessages(foundOldest: true, foundNewest: false,
+      messages: List.generate(30, (i) => eg.streamMessage(stream: stream)));
 
     check(model).messages.length.equals(30);
     final otherStream = eg.stream();
@@ -380,7 +542,7 @@ void main() {
       final otherStream = eg.stream();
       await store.addStream(otherStream);
       await store.addSubscription(eg.subscription(otherStream));
-      await prepareMessages(foundOldest: true, messages: [
+      await prepareMessages(foundOldest: true, foundNewest: false, messages: [
         eg.streamMessage(id: 1, stream: stream, topic: 'bar'),
         eg.streamMessage(id: 2, stream: stream, topic: topic),
         eg.streamMessage(id: 3, stream: otherStream, topic: 'elsewhere'),
@@ -398,7 +560,7 @@ void main() {
       await prepare(narrow: const CombinedFeedNarrow());
       // Mute the stream, so that combined-feed vs. stream visibility differ.
       await prepareMutes(true, UserTopicVisibilityPolicy.followed);
-      await prepareMessages(foundOldest: true, messages: [
+      await prepareMessages(foundOldest: true, foundNewest: false, messages: [
         eg.streamMessage(id: 1, stream: stream, topic: topic),
       ]);
       checkHasMessageIds([1]);
@@ -421,7 +583,7 @@ void main() {
       await prepare(narrow: ChannelNarrow(stream.streamId));
       // Mute the stream, so that combined-feed vs. stream visibility differ.
       await prepareMutes(true, UserTopicVisibilityPolicy.followed);
-      await prepareMessages(foundOldest: true, messages: [
+      await prepareMessages(foundOldest: true, foundNewest: false, messages: [
         eg.streamMessage(id: 1, stream: stream, topic: topic),
       ]);
       checkHasMessageIds([1]);
@@ -442,7 +604,7 @@ void main() {
     test('in TopicNarrow, stay visible', () async {
       await prepare(narrow: eg.topicNarrow(stream.streamId, topic));
       await prepareMutes();
-      await prepareMessages(foundOldest: true, messages: [
+      await prepareMessages(foundOldest: true, foundNewest: false, messages: [
         eg.streamMessage(id: 1, stream: stream, topic: topic),
       ]);
       checkHasMessageIds([1]);
@@ -456,7 +618,7 @@ void main() {
       await prepare(narrow:
         DmNarrow.withUser(eg.otherUser.userId, selfUserId: eg.selfUser.userId));
       await prepareMutes();
-      await prepareMessages(foundOldest: true, messages: [
+      await prepareMessages(foundOldest: true, foundNewest: false, messages: [
         eg.dmMessage(id: 1, from: eg.otherUser, to: [eg.selfUser]),
       ]);
       checkHasMessageIds([1]);
@@ -469,7 +631,7 @@ void main() {
     test('no affected messages -> no notification', () async {
       await prepare(narrow: const CombinedFeedNarrow());
       await prepareMutes();
-      await prepareMessages(foundOldest: true, messages: [
+      await prepareMessages(foundOldest: true, foundNewest: false, messages: [
         eg.streamMessage(id: 1, stream: stream, topic: 'bar'),
       ]);
       checkHasMessageIds([1]);
@@ -486,11 +648,11 @@ void main() {
         eg.dmMessage(id: 1, from: eg.otherUser, to: [eg.selfUser]),
         eg.streamMessage(id: 2, stream: stream, topic: topic),
       ];
-      await prepareMessages(foundOldest: true, messages: messages);
+      await prepareMessages(foundOldest: true, foundNewest: false, messages: messages);
       checkHasMessageIds([1]);
 
       connection.prepare(
-        json: newestResult(foundOldest: true, messages: messages).toJson());
+        json: nearUnreadResult(foundOldest: true, foundNewest: false, messages: messages).toJson());
       await setVisibility(UserTopicVisibilityPolicy.unmuted);
       checkNotifiedOnce();
       check(model).fetched.isFalse();
@@ -509,7 +671,7 @@ void main() {
       ];
 
       connection.prepare(
-        json: newestResult(foundOldest: true, messages: messages).toJson());
+        json: nearUnreadResult(foundOldest: true, foundNewest: false, messages: messages).toJson());
       final fetchFuture = model.fetchInitial();
 
       await setVisibility(UserTopicVisibilityPolicy.unmuted);
@@ -528,7 +690,7 @@ void main() {
 
     test('in narrow', () async {
       await prepare(narrow: ChannelNarrow(stream.streamId));
-      await prepareMessages(foundOldest: true, messages: messages);
+      await prepareMessages(foundOldest: true, foundNewest: false, messages: messages);
 
       check(model).messages.length.equals(30);
       await store.handleEvent(eg.deleteMessageEvent(messages.sublist(0, 10)));
@@ -538,7 +700,7 @@ void main() {
 
     test('not all in narrow', () async {
       await prepare(narrow: ChannelNarrow(stream.streamId));
-      await prepareMessages(foundOldest: true, messages: messages.sublist(5));
+      await prepareMessages(foundOldest: true, foundNewest: false, messages: messages.sublist(5));
 
       check(model).messages.length.equals(25);
       await store.handleEvent(eg.deleteMessageEvent(messages.sublist(0, 10)));
@@ -548,7 +710,7 @@ void main() {
 
     test('not in narrow', () async {
       await prepare(narrow: ChannelNarrow(stream.streamId));
-      await prepareMessages(foundOldest: true, messages: messages.sublist(5));
+      await prepareMessages(foundOldest: true, foundNewest: false, messages: messages.sublist(5));
 
       check(model).messages.length.equals(25);
       await store.handleEvent(eg.deleteMessageEvent(messages.sublist(0, 5)));
@@ -558,7 +720,7 @@ void main() {
 
     test('complete message deletion', () async {
       await prepare(narrow: ChannelNarrow(stream.streamId));
-      await prepareMessages(foundOldest: true, messages: messages.sublist(0, 25));
+      await prepareMessages(foundOldest: true, foundNewest: false, messages: messages.sublist(0, 25));
 
       check(model).messages.length.equals(25);
       await store.handleEvent(eg.deleteMessageEvent(messages));
@@ -568,7 +730,7 @@ void main() {
 
     test('non-consecutive message deletion', () async {
       await prepare(narrow: ChannelNarrow(stream.streamId));
-      await prepareMessages(foundOldest: true, messages: messages);
+      await prepareMessages(foundOldest: true, foundNewest: false, messages: messages);
       final messagesToDelete = messages.sublist(2, 5) + messages.sublist(10, 15);
 
       check(model).messages.length.equals(30);
@@ -585,7 +747,7 @@ void main() {
   group('notifyListenersIfMessagePresent', () {
     test('message present', () async {
       await prepare(narrow: const CombinedFeedNarrow());
-      await prepareMessages(foundOldest: false,
+      await prepareMessages(foundOldest: false, foundNewest: false,
         messages: List.generate(100, (i) => eg.streamMessage(id: 100 + i)));
       model.notifyListenersIfMessagePresent(150);
       checkNotifiedOnce();
@@ -593,7 +755,7 @@ void main() {
 
     test('message absent', () async {
       await prepare(narrow: const CombinedFeedNarrow());
-      await prepareMessages(foundOldest: false,
+      await prepareMessages(foundOldest: false, foundNewest: false,
         messages: List.generate(100, (i) => eg.streamMessage(id: 100 + i))
           .where((m) => m.id != 150).toList());
       model.notifyListenersIfMessagePresent(150);
@@ -602,7 +764,7 @@ void main() {
 
     test('message absent (older than window)', () async {
       await prepare(narrow: const CombinedFeedNarrow());
-      await prepareMessages(foundOldest: false,
+      await prepareMessages(foundOldest: false, foundNewest: false,
         messages: List.generate(100, (i) => eg.streamMessage(id: 100 + i)));
       model.notifyListenersIfMessagePresent(50);
       checkNotNotified();
@@ -610,7 +772,7 @@ void main() {
 
     test('message absent (newer than window)', () async {
       await prepare(narrow: const CombinedFeedNarrow());
-      await prepareMessages(foundOldest: false,
+      await prepareMessages(foundOldest: false, foundNewest: false,
         messages: List.generate(100, (i) => eg.streamMessage(id: 100 + i)));
       model.notifyListenersIfMessagePresent(250);
       checkNotNotified();
@@ -622,14 +784,14 @@ void main() {
 
     test('all messages present', () async {
       await prepare(narrow: const CombinedFeedNarrow());
-      await prepareMessages(foundOldest: false, messages: messages);
+      await prepareMessages(foundOldest: false, foundNewest: false, messages: messages);
       model.notifyListenersIfAnyMessagePresent([150, 151, 152]);
       checkNotifiedOnce();
     });
 
     test('some messages present', () async {
       await prepare(narrow: const CombinedFeedNarrow());
-      await prepareMessages(foundOldest: false,
+      await prepareMessages(foundOldest: false, foundNewest: false,
         messages: messages.where((m) => m.id != 151).toList());
       model.notifyListenersIfAnyMessagePresent([150, 151, 152]);
       checkNotifiedOnce();
@@ -637,8 +799,8 @@ void main() {
 
     test('no messages present', () async {
       await prepare(narrow: const CombinedFeedNarrow());
-      await prepareMessages(foundOldest: false, messages:
-        messages.where((m) => ![150, 151, 152].contains(m.id)).toList());
+      await prepareMessages(foundOldest: false, foundNewest: false,
+        messages: messages.where((m) => ![150, 151, 152].contains(m.id)).toList());
       model.notifyListenersIfAnyMessagePresent([150, 151, 152]);
       checkNotNotified();
     });
@@ -647,7 +809,7 @@ void main() {
   group('messageContentChanged', () {
     test('message present', () async {
       await prepare(narrow: const CombinedFeedNarrow());
-      await prepareMessages(foundOldest: false,
+      await prepareMessages(foundOldest: false, foundNewest: false,
         messages: List.generate(10, (i) => eg.streamMessage(id: 10 + i)));
 
       final message = model.messages[5];
@@ -668,7 +830,7 @@ void main() {
       final messageNotInNarrow = eg.dmMessage(id: 100, from: eg.otherUser, to: [eg.selfUser]);
       check(narrow.containsMessage(messageNotInNarrow)).isFalse();
 
-      await prepareMessages(foundOldest: false, messages: messagesInNarrow);
+      await prepareMessages(foundOldest: false, foundNewest: false, messages: messagesInNarrow);
       await store.addMessage(messageNotInNarrow);
 
       await store.handleEvent(eg.updateMessageEditEvent(messageNotInNarrow,
@@ -693,7 +855,7 @@ void main() {
         await store.addSubscription(subscription);
       }
       if (messages != null) {
-        await prepareMessages(foundOldest: false, messages: messages);
+        await prepareMessages(foundOldest: false, foundNewest: false, messages: messages);
       }
       checkHasMessages(messages ?? []);
     }
@@ -747,8 +909,9 @@ void main() {
       test('old channel -> channel: refetch', () => awaitFakeAsync((async) async {
         await prepareNarrow(narrow, initialMessages);
 
-        connection.prepare(delay: const Duration(seconds: 2), json: newestResult(
+        connection.prepare(delay: const Duration(seconds: 2), json: nearUnreadResult(
           foundOldest: false,
+          foundNewest: false,
           messages: initialMessages + movedMessages,
         ).toJson());
         await store.handleEvent(eg.updateMessageEventMoveTo(
@@ -801,8 +964,9 @@ void main() {
 
       void testMessageMove(PropagateMode propagateMode) => awaitFakeAsync((async) async {
         await prepareNarrow(narrow, initialMessages + movedMessages);
-        connection.prepare(delay: const Duration(seconds: 1), json: newestResult(
+        connection.prepare(delay: const Duration(seconds: 1), json: nearUnreadResult(
           foundOldest: false,
+          foundNewest: false,
           messages: movedMessages,
         ).toJson());
         await store.handleEvent(eg.updateMessageEventMoveFrom(
@@ -849,8 +1013,9 @@ void main() {
           test(description, () => awaitFakeAsync((async) async {
             await prepareNarrow(narrow, initialMessages);
 
-            connection.prepare(delay: const Duration(seconds: 2), json: newestResult(
+            connection.prepare(delay: const Duration(seconds: 2), json: nearUnreadResult(
               foundOldest: false,
+              foundNewest: false,
               messages: initialMessages + movedMessages,
             ).toJson());
             await store.handleEvent(eg.updateMessageEventMoveTo(
@@ -919,8 +1084,9 @@ void main() {
 
       void handleMoveEvent(PropagateMode propagateMode) => awaitFakeAsync((async) async {
         await prepareNarrow(narrow, initialMessages + movedMessages);
-        connection.prepare(delay: const Duration(seconds: 1), json: newestResult(
+        connection.prepare(delay: const Duration(seconds: 1), json: nearUnreadResult(
           foundOldest: false,
+          foundNewest: false,
           messages: movedMessages,
         ).toJson());
         await store.handleEvent(eg.updateMessageEventMoveFrom(
@@ -961,8 +1127,9 @@ void main() {
         await store.addSubscription(subscription);
         final followedMessage = eg.streamMessage(stream: stream, topic: 'new');
 
-        connection.prepare(delay: const Duration(seconds: 2), json: newestResult(
+        connection.prepare(delay: const Duration(seconds: 2), json: nearUnreadResult(
           foundOldest: true,
+          foundNewest: false,
           messages: [followedMessage],
         ).toJson());
 
@@ -999,8 +1166,9 @@ void main() {
         checkHasMessages(initialMessages);
         checkNotifiedOnce();
 
-        connection.prepare(delay: const Duration(seconds: 2), json: newestResult(
+        connection.prepare(delay: const Duration(seconds: 2), json: nearUnreadResult(
           foundOldest: false,
+          foundNewest: false,
           messages: initialMessages + movedMessages,
         ).toJson());
         await store.handleEvent(eg.updateMessageEventMoveTo(
@@ -1034,8 +1202,9 @@ void main() {
         check(model).fetchingOlder.isTrue();
         checkNotifiedOnce();
 
-        connection.prepare(delay: const Duration(seconds: 1), json: newestResult(
+        connection.prepare(delay: const Duration(seconds: 1), json: nearUnreadResult(
           foundOldest: false,
+          foundNewest: false,
           messages: initialMessages + movedMessages,
         ).toJson());
         await store.handleEvent(eg.updateMessageEventMoveTo(
@@ -1071,8 +1240,9 @@ void main() {
         checkHasMessages(initialMessages);
         checkNotified(count: 2);
 
-        connection.prepare(json: newestResult(
+        connection.prepare(json: nearUnreadResult(
           foundOldest: false,
+          foundNewest: false,
           messages: initialMessages + movedMessages,
         ).toJson());
         await store.handleEvent(eg.updateMessageEventMoveTo(
@@ -1123,16 +1293,18 @@ void main() {
       test('fetchInitial, _reset, initial fetch finishes, move fetch finishes', () => awaitFakeAsync((async) async {
         await prepareNarrow(narrow, null);
 
-        connection.prepare(delay: const Duration(seconds: 1), json: newestResult(
+        connection.prepare(delay: const Duration(seconds: 1), json: nearUnreadResult(
           foundOldest: false,
+          foundNewest: false,
           messages: initialMessages,
         ).toJson());
         final fetchFuture = model.fetchInitial();
         checkHasMessages([]);
         check(model).fetched.isFalse();
 
-        connection.prepare(delay: const Duration(seconds: 2), json: newestResult(
+        connection.prepare(delay: const Duration(seconds: 2), json: nearUnreadResult(
           foundOldest: false,
+          foundNewest: false,
           messages: initialMessages + movedMessages,
         ).toJson());
         await store.handleEvent(eg.updateMessageEventMoveTo(
@@ -1157,16 +1329,18 @@ void main() {
       test('fetchInitial, _reset, move fetch finishes, initial fetch finishes', () => awaitFakeAsync((async) async {
         await prepareNarrow(narrow, null);
 
-        connection.prepare(delay: const Duration(seconds: 2), json: newestResult(
+        connection.prepare(delay: const Duration(seconds: 2), json: nearUnreadResult(
           foundOldest: false,
+          foundNewest: false,
           messages: initialMessages,
         ).toJson());
         final fetchFuture = model.fetchInitial();
         checkHasMessages([]);
         check(model).fetched.isFalse();
 
-        connection.prepare(delay: const Duration(seconds: 1), json: newestResult(
+        connection.prepare(delay: const Duration(seconds: 1), json: nearUnreadResult(
           foundOldest: false,
+          foundNewest: false,
           messages: initialMessages + movedMessages,
         ).toJson());
         await store.handleEvent(eg.updateMessageEventMoveTo(
@@ -1199,8 +1373,9 @@ void main() {
         check(model).fetchingOlder.isTrue();
         checkNotifiedOnce();
 
-        connection.prepare(delay: const Duration(seconds: 1), json: newestResult(
+        connection.prepare(delay: const Duration(seconds: 1), json: nearUnreadResult(
           foundOldest: false,
+          foundNewest: false,
           messages: initialMessages + movedMessages,
         ).toJson());
         await store.handleEvent(eg.updateMessageEventMoveTo(
@@ -1259,8 +1434,9 @@ void main() {
         ..addListener(() => notifiedCount2++);
 
       for (final m in [model1, model2]) {
-        connection.prepare(json: newestResult(
+        connection.prepare(json: nearUnreadResult(
           foundOldest: false,
+          foundNewest: false,
           messages: [eg.streamMessage(stream: stream, topic: 'hello')]).toJson());
         await m.fetchInitial();
       }
@@ -1299,7 +1475,7 @@ void main() {
       checkInvariants(model);
 
       connection.prepare(json:
-        newestResult(foundOldest: false, messages: [message]).toJson());
+        nearUnreadResult(foundOldest: false, foundNewest: false, messages: [message]).toJson());
       await model.fetchInitial();
       checkInvariants(model);
       doCheckMessageAfterFetch(
@@ -1346,7 +1522,7 @@ void main() {
   test('reassemble', () async {
     final stream = eg.stream();
     await prepare(narrow: ChannelNarrow(stream.streamId));
-    await prepareMessages(foundOldest: true, messages:
+    await prepareMessages(foundOldest: true, foundNewest: false, messages:
       List.generate(30, (i) => eg.streamMessage(stream: stream)));
     await store.handleEvent(MessageEvent(id: 0,
       message: eg.streamMessage(stream: stream)));
@@ -1380,7 +1556,7 @@ void main() {
       await store.addUserTopic(stream2, 'C', UserTopicVisibilityPolicy.unmuted);
 
       // Check filtering on fetchInitial…
-      await prepareMessages(foundOldest: false, messages: [
+      await prepareMessages(foundOldest: false, foundNewest: false, messages: [
         eg.streamMessage(id: 201, stream: stream1, topic: 'A'),
         eg.streamMessage(id: 202, stream: stream1, topic: 'B'),
         eg.streamMessage(id: 203, stream: stream2, topic: 'C'),
@@ -1441,7 +1617,7 @@ void main() {
       await store.addUserTopic(stream, 'C', UserTopicVisibilityPolicy.muted);
 
       // Check filtering on fetchInitial…
-      await prepareMessages(foundOldest: false, messages: [
+      await prepareMessages(foundOldest: false, foundNewest: false, messages: [
         eg.streamMessage(id: 201, stream: stream, topic: 'A'),
         eg.streamMessage(id: 202, stream: stream, topic: 'B'),
         eg.streamMessage(id: 203, stream: stream, topic: 'C'),
@@ -1487,7 +1663,7 @@ void main() {
       await store.addUserTopic(stream, 'A', UserTopicVisibilityPolicy.muted);
 
       // Check filtering on fetchInitial…
-      await prepareMessages(foundOldest: false, messages: [
+      await prepareMessages(foundOldest: false, foundNewest: false, messages: [
         eg.streamMessage(id: 201, stream: stream, topic: 'A'),
       ]);
       final expected = <int>[];
@@ -1529,7 +1705,7 @@ void main() {
       ];
 
       // Check filtering on fetchInitial…
-      await prepareMessages(foundOldest: false, messages: getMessages(201));
+      await prepareMessages(foundOldest: false, foundNewest: false, messages: getMessages(201));
       final expected = <int>[];
       check(model.messages.map((m) => m.id))
         .deepEquals(expected..addAll([201, 202, 203]));
@@ -1567,7 +1743,7 @@ void main() {
       ];
 
       // Check filtering on fetchInitial…
-      await prepareMessages(foundOldest: false, messages: getMessages(201));
+      await prepareMessages(foundOldest: false, foundNewest: false, messages: getMessages(201));
       final expected = <int>[];
       check(model.messages.map((m) => m.id))
         .deepEquals(expected..addAll([201, 202]));
@@ -1594,7 +1770,7 @@ void main() {
     test('ZulipContent', () async {
       final stream = eg.stream();
       await prepare(narrow: ChannelNarrow(stream.streamId));
-      await prepareMessages(foundOldest: true, messages: []);
+      await prepareMessages(foundOldest: true, foundNewest: false, messages: []);
 
       await store.handleEvent(MessageEvent(id: 0,
         message: eg.streamMessage(stream: stream)));
@@ -1609,7 +1785,7 @@ void main() {
     test('PollContent', () async {
       final stream = eg.stream();
       await prepare(narrow: ChannelNarrow(stream.streamId));
-      await prepareMessages(foundOldest: true, messages: []);
+      await prepareMessages(foundOldest: true, foundNewest: false, messages: []);
 
       await store.handleEvent(MessageEvent(id: 0, message: eg.streamMessage(
         stream: stream,
@@ -1649,8 +1825,9 @@ void main() {
 
     // First, test fetchInitial, where some headers are needed and others not.
     await prepare();
-    connection.prepare(json: newestResult(
+    connection.prepare(json: nearUnreadResult(
       foundOldest: false,
+      foundNewest: false,
       messages: [streamMessage(10), streamMessage(11), dmMessage(12)],
     ).toJson());
     await model.fetchInitial();
@@ -1731,7 +1908,7 @@ void main() {
         to: [sender.userId == eg.selfUser.userId ? eg.otherUser : eg.selfUser]);
 
     await prepare();
-    await prepareMessages(foundOldest: true, messages: [
+    await prepareMessages(foundOldest: true, foundNewest: false, messages: [
       streamMessage(1, t1, eg.selfUser),  // first message, so show sender
       streamMessage(2, t1, eg.selfUser),  // hide sender
       streamMessage(3, t1, eg.otherUser), // no recipient header, but new sender
@@ -1871,7 +2048,10 @@ void checkInvariants(MessageListView model) {
       ..messages.isEmpty()
       ..haveOldest.isFalse()
       ..fetchingOlder.isFalse()
-      ..fetchOlderCoolingDown.isFalse();
+      ..fetchOlderCoolingDown.isFalse()
+      ..haveNewest.isFalse()
+      ..fetchingNewer.isFalse()
+      ..fetchNewerCoolingDown.isFalse();
   }
   if (model.haveOldest) {
     check(model).fetchingOlder.isFalse();
@@ -1880,6 +2060,13 @@ void checkInvariants(MessageListView model) {
   if (model.fetchingOlder) {
     check(model).fetchOlderCoolingDown.isFalse();
   }
+  if (model.haveNewest) {
+    check(model).fetchingNewer.isFalse();
+    check(model).fetchNewerCoolingDown.isFalse();
+  }
+  if (model.fetchingNewer) {
+    check(model).fetchNewerCoolingDown.isFalse();
+  }
 
   for (final message in model.messages) {
     check(model.store.messages)[message.id].isNotNull().identicalTo(message);
@@ -1947,6 +2134,9 @@ void checkInvariants(MessageListView model) {
           || MessageListLoadingItem()       => true,
         });
   }
+  if (model.fetchingNewer || model.fetchNewerCoolingDown) {
+    check(model.items[i++]).isA<MessageListLoadingItem>();
+  }
   check(model.items).length.equals(i);
 }
 
@@ -1975,4 +2165,7 @@ extension MessageListViewChecks on Subject<MessageListView> {
   Subject<bool> get haveOldest => has((x) => x.haveOldest, 'haveOldest');
   Subject<bool> get fetchingOlder => has((x) => x.fetchingOlder, 'fetchingOlder');
   Subject<bool> get fetchOlderCoolingDown => has((x) => x.fetchOlderCoolingDown, 'fetchOlderCoolingDown');
+  Subject<bool> get haveNewest => has((x) => x.haveNewest, 'haveNewest');
+  Subject<bool> get fetchingNewer => has((x) => x.fetchingNewer, 'fetchingNewer');
+  Subject<bool> get fetchNewerCoolingDown => has((x) => x.fetchNewerCoolingDown, 'fetchNewerCoolingDown');
 }
diff --git a/test/model/message_test.dart b/test/model/message_test.dart
index 43f17be61a..daa04e884f 100644
--- a/test/model/message_test.dart
+++ b/test/model/message_test.dart
@@ -62,10 +62,14 @@ void main() {
   Future<void> prepareMessages(
     List<Message> messages, {
     bool foundOldest = false,
+    bool foundNewest = false,
   }) async {
     assert(messages.every((message) => message.poll == null));
     connection.prepare(json:
-      eg.newestGetMessagesResult(foundOldest: foundOldest, messages: messages).toJson());
+      eg.nearUnreadGetMessagesResult(
+        foundOldest: foundOldest,
+        foundNewest: foundNewest,
+        messages: messages).toJson());
     await messageList.fetchInitial();
     checkNotifiedOnce();
   }
@@ -645,7 +649,7 @@ void main() {
         // Perform a single-message initial message fetch for [messageList] with
         // submessages.
         connection.prepare(json:
-          eg.newestGetMessagesResult(foundOldest: true, messages: []).toJson()
+          eg.nearUnreadGetMessagesResult(foundOldest: true, foundNewest: true, messages: []).toJson()
             ..['messages'] = [{
               ...message.toJson(),
               "submessages": submessages.map(deepToJson).toList(),
diff --git a/test/widgets/action_sheet_test.dart b/test/widgets/action_sheet_test.dart
index c10957363e..677cf2efa9 100644
--- a/test/widgets/action_sheet_test.dart
+++ b/test/widgets/action_sheet_test.dart
@@ -67,8 +67,8 @@ Future<void> setupToMessageActionSheet(WidgetTester tester, {
   }
   connection = store.connection as FakeApiConnection;
 
-  connection.prepare(json: eg.newestGetMessagesResult(
-    foundOldest: true, messages: [message]).toJson());
+  connection.prepare(json: eg.nearUnreadGetMessagesResult(
+    foundOldest: true, foundNewest: true, messages: [message]).toJson());
   await tester.pumpWidget(TestZulipApp(accountId: eg.selfAccount.id,
     child: MessageListPage(initNarrow: narrow)));
 
@@ -141,8 +141,8 @@ void main() {
 
     testWidgets('show from app bar', (tester) async {
       await prepare();
-      connection.prepare(json: eg.newestGetMessagesResult(
-        foundOldest: true, messages: [message]).toJson());
+      connection.prepare(json: eg.nearUnreadGetMessagesResult(
+        foundOldest: true, foundNewest: true, messages: [message]).toJson());
       await tester.pumpWidget(TestZulipApp(accountId: eg.selfAccount.id,
         child: MessageListPage(
           initNarrow: eg.topicNarrow(channel.streamId, topic))));
@@ -158,8 +158,8 @@ void main() {
 
     testWidgets('show from recipient header', (tester) async {
       await prepare();
-      connection.prepare(json: eg.newestGetMessagesResult(
-        foundOldest: true, messages: [message]).toJson());
+      connection.prepare(json: eg.nearUnreadGetMessagesResult(
+        foundOldest: false, foundNewest: false, messages: [message]).toJson());
       await tester.pumpWidget(TestZulipApp(accountId: eg.selfAccount.id,
         child: const MessageListPage(initNarrow: CombinedFeedNarrow())));
       // global store, per-account store, and message list get loaded
@@ -209,8 +209,8 @@ void main() {
       store = await testBinding.globalStore.perAccount(account.id);
       connection = store.connection as FakeApiConnection;
 
-      connection.prepare(json: eg.newestGetMessagesResult(
-        foundOldest: true, messages: [
+      connection.prepare(json: eg.nearUnreadGetMessagesResult(
+        foundOldest: true, foundNewest: true, messages: [
           eg.streamMessage(stream: channel, topic: topic)]).toJson());
       await tester.pumpWidget(TestZulipApp(accountId: account.id,
         child: MessageListPage(
@@ -806,8 +806,8 @@ void main() {
         // it doesn't matter anyway: [MessageStoreImpl.reconcileMessages] will
         // keep the version updated by the event.  If that somehow changes in
         // some future refactor, it'll cause this test to fail.
-        connection.prepare(json: eg.newestGetMessagesResult(
-          foundOldest: true, messages: [message]).toJson());
+        connection.prepare(json: eg.nearUnreadGetMessagesResult(
+          foundOldest: true, foundNewest: true, messages: [message]).toJson());
         await store.handleEvent(eg.updateMessageEventMoveFrom(
           newStreamId: newStream.streamId, newTopicStr: newTopic,
           propagateMode: PropagateMode.changeAll,
diff --git a/test/widgets/emoji_reaction_test.dart b/test/widgets/emoji_reaction_test.dart
index 89333ee1af..440eee46ed 100644
--- a/test/widgets/emoji_reaction_test.dart
+++ b/test/widgets/emoji_reaction_test.dart
@@ -324,8 +324,8 @@ void main() {
       await store.addSubscription(eg.subscription(stream));
 
       connection = store.connection as FakeApiConnection;
-      connection.prepare(json: eg.newestGetMessagesResult(
-        foundOldest: true, messages: [message]).toJson());
+      connection.prepare(json: eg.nearUnreadGetMessagesResult(
+        foundOldest: true, foundNewest: true, messages: [message]).toJson());
       await tester.pumpWidget(TestZulipApp(accountId: eg.selfAccount.id,
         child: MessageListPage(initNarrow: narrow)));
 
diff --git a/test/widgets/home_test.dart b/test/widgets/home_test.dart
index 5bb789727f..67a832716f 100644
--- a/test/widgets/home_test.dart
+++ b/test/widgets/home_test.dart
@@ -125,8 +125,8 @@ void main () {
       await prepare(tester, navigatorObserver: testNavObserver);
       pushedRoutes.clear();
 
-      connection.prepare(json: eg.newestGetMessagesResult(
-        foundOldest: true, messages: []).toJson());
+      connection.prepare(json: eg.nearUnreadGetMessagesResult(
+        foundOldest: true, foundNewest: true, messages: []).toJson());
       await tester.tap(find.byIcon(ZulipIcons.message_feed));
       await tester.pump();
       await tester.pump(const Duration(milliseconds: 250));
@@ -236,8 +236,8 @@ void main () {
 
       await tapOpenMenu(tester);
 
-      connection.prepare(json: eg.newestGetMessagesResult(
-        foundOldest: true, messages: [eg.streamMessage()]).toJson());
+      connection.prepare(json: eg.nearUnreadGetMessagesResult(
+        foundOldest: true, foundNewest: true, messages: [eg.streamMessage()]).toJson());
       await tester.tap(combinedFeedMenuIconFinder);
       await tester.pump(Duration.zero); // tap the button
       await tester.pump(const Duration(milliseconds: 250)); // wait for animation
diff --git a/test/widgets/message_list_test.dart b/test/widgets/message_list_test.dart
index aadb2ffc77..1ee5040b75 100644
--- a/test/widgets/message_list_test.dart
+++ b/test/widgets/message_list_test.dart
@@ -74,7 +74,10 @@ void main() {
       return eg.streamMessage(sender: eg.selfUser);
     });
     connection.prepare(json:
-      eg.newestGetMessagesResult(foundOldest: foundOldest, messages: messages).toJson());
+      eg.nearUnreadGetMessagesResult(
+        foundOldest: foundOldest,
+        foundNewest: true,
+        messages: messages).toJson());
 
     await tester.pumpWidget(TestZulipApp(accountId: eg.selfAccount.id,
       navigatorObservers: navObservers,
@@ -664,8 +667,8 @@ void main() {
     final narrow = eg.topicNarrow(channel.streamId, topic);
 
     void prepareGetMessageResponse(List<Message> messages) {
-      connection.prepare(json: eg.newestGetMessagesResult(
-        foundOldest: false, messages: messages).toJson());
+      connection.prepare(json: eg.nearUnreadGetMessagesResult(
+        foundOldest: false, foundNewest: false, messages: messages).toJson());
     }
 
     void handleMessageMoveEvent(List<StreamMessage> messages, String newTopic, {int? newChannelId}) {