diff --git a/stubs/taint/false_positives.pysa b/stubs/taint/false_positives.pysa
index ce4d5715bb..f57cbd1b0c 100644
--- a/stubs/taint/false_positives.pysa
+++ b/stubs/taint/false_positives.pysa
@@ -16,9 +16,9 @@ def zerver.lib.thumbnail.generate_thumbnail_url(
 # filesystem operations.
 def zerver.lib.upload.sanitize_name(value) -> Sanitize: ...
 
-# This function accepts two integers and then concatenates them into a path
+# This function accepts three integers and then concatenates them into a path
 # segment. The result should be safe for use in filesystem and other operations.
-def zerver.lib.avatar_hash.user_avatar_path_from_ids(user_profile_id, realm_id) -> Sanitize: ...
+def zerver.lib.avatar_hash.user_avatar_base_path_from_ids(user_profile_id, version, realm_id) -> Sanitize: ...
 
 # This function creates a list of 'UserMessageLite' objects, which contain only
 # integral IDs and flags. These should safe for use with SQL and other
diff --git a/zerver/actions/user_settings.py b/zerver/actions/user_settings.py
index d16ad66432..77c218cd38 100644
--- a/zerver/actions/user_settings.py
+++ b/zerver/actions/user_settings.py
@@ -397,8 +397,9 @@ def do_change_avatar_fields(
 
 
 def do_delete_avatar_image(user: UserProfile, *, acting_user: Optional[UserProfile]) -> None:
+    old_version = user.avatar_version
     do_change_avatar_fields(user, UserProfile.AVATAR_FROM_GRAVATAR, acting_user=acting_user)
-    delete_avatar_image(user)
+    delete_avatar_image(user, old_version)
 
 
 def update_scheduled_email_notifications_time(
diff --git a/zerver/data_import/import_util.py b/zerver/data_import/import_util.py
index 7fde75a8ec..7cc5cb7377 100644
--- a/zerver/data_import/import_util.py
+++ b/zerver/data_import/import_util.py
@@ -27,7 +27,7 @@ from django.utils.timezone import now as timezone_now
 from typing_extensions import TypeAlias
 
 from zerver.data_import.sequencer import NEXT_ID
-from zerver.lib.avatar_hash import user_avatar_path_from_ids
+from zerver.lib.avatar_hash import user_avatar_base_path_from_ids
 from zerver.lib.partial import partial
 from zerver.lib.stream_color import STREAM_ASSIGNMENT_COLORS as STREAM_COLORS
 from zerver.models import (
@@ -148,6 +148,7 @@ def build_avatar(
         path=avatar_url,  # Save original avatar URL here, which is downloaded later
         realm_id=realm_id,
         content_type=None,
+        avatar_version=1,
         user_profile_id=zulip_user_id,
         last_modified=timestamp,
         user_profile_email=email,
@@ -594,7 +595,9 @@ def process_avatars(
     avatar_original_list = []
     avatar_upload_list = []
     for avatar in avatar_list:
-        avatar_hash = user_avatar_path_from_ids(avatar["user_profile_id"], realm_id)
+        avatar_hash = user_avatar_base_path_from_ids(
+            avatar["user_profile_id"], avatar["avatar_version"], realm_id
+        )
         avatar_url = avatar["path"]
         avatar_original = dict(avatar)
 
diff --git a/zerver/lib/avatar.py b/zerver/lib/avatar.py
index d23b579bc4..6abec2bd4c 100644
--- a/zerver/lib/avatar.py
+++ b/zerver/lib/avatar.py
@@ -6,8 +6,8 @@ from django.contrib.staticfiles.storage import staticfiles_storage
 
 from zerver.lib.avatar_hash import (
     gravatar_hash,
+    user_avatar_base_path_from_ids,
     user_avatar_content_hash,
-    user_avatar_path_from_ids,
 )
 from zerver.lib.thumbnail import MEDIUM_AVATAR_SIZE
 from zerver.lib.upload import get_avatar_url
@@ -55,28 +55,28 @@ def get_avatar_field(
             computing them on the server (mostly to save bandwidth).
     """
 
-    if client_gravatar:
-        """
-        If our client knows how to calculate gravatar hashes, we
-        will return None and let the client compute the gravatar
-        url.
-        """
-        if settings.ENABLE_GRAVATAR and avatar_source == UserProfile.AVATAR_FROM_GRAVATAR:
-            return None
+    """
+    If our client knows how to calculate gravatar hashes, we
+    will return None and let the client compute the gravatar
+    url.
+    """
+    if (
+        client_gravatar
+        and settings.ENABLE_GRAVATAR
+        and avatar_source == UserProfile.AVATAR_FROM_GRAVATAR
+    ):
+        return None
 
     """
     If we get this far, we'll compute an avatar URL that may be
     either user-uploaded or a gravatar, and then we'll add version
     info to try to avoid stale caches.
     """
-    url = _get_unversioned_avatar_url(
-        user_profile_id=user_id,
-        avatar_source=avatar_source,
-        realm_id=realm_id,
-        email=email,
-        medium=medium,
-    )
-    return append_url_query_string(url, f"version={avatar_version:d}")
+    if avatar_source == "U":
+        hash_key = user_avatar_base_path_from_ids(user_id, avatar_version, realm_id)
+        return get_avatar_url(hash_key, medium=medium)
+
+    return get_gravatar_url(email=email, avatar_version=avatar_version, medium=medium)
 
 
 def get_gravatar_url(email: str, avatar_version: int, medium: bool = False) -> str:
@@ -95,20 +95,6 @@ def _get_unversioned_gravatar_url(email: str, medium: bool) -> str:
         return staticfiles_storage.url("images/default-avatar.png")
 
 
-def _get_unversioned_avatar_url(
-    user_profile_id: int,
-    avatar_source: str,
-    realm_id: int,
-    email: Optional[str] = None,
-    medium: bool = False,
-) -> str:
-    if avatar_source == "U":
-        hash_key = user_avatar_path_from_ids(user_profile_id, realm_id)
-        return get_avatar_url(hash_key, medium=medium)
-    assert email is not None
-    return _get_unversioned_gravatar_url(email, medium)
-
-
 def absolute_avatar_url(user_profile: UserProfile) -> str:
     """
     Absolute URLs are used to simplify logic for applications that
diff --git a/zerver/lib/avatar_hash.py b/zerver/lib/avatar_hash.py
index c500c12173..e2bfcf364b 100644
--- a/zerver/lib/avatar_hash.py
+++ b/zerver/lib/avatar_hash.py
@@ -15,25 +15,25 @@ def gravatar_hash(email: str) -> str:
     return hashlib.md5(email.lower().encode()).hexdigest()
 
 
-def user_avatar_hash(uid: str) -> str:
+def user_avatar_hash(uid: str, version: str) -> str:
     # WARNING: If this method is changed, you may need to do a migration
     # similar to zerver/migrations/0060_move_avatars_to_be_uid_based.py .
 
-    # The salt probably doesn't serve any purpose now.  In the past we
-    # used a hash of the email address, not the user ID, and we salted
-    # it in order to make the hashing scheme different from Gravatar's.
-    user_key = uid + settings.AVATAR_SALT
-    return hashlib.sha1(user_key.encode()).hexdigest()
+    # The salt prevents unauthenticated clients from enumerating the
+    # avatars of all users.
+    user_key = uid + ":" + version + ":" + settings.AVATAR_SALT
+    return hashlib.sha256(user_key.encode()).hexdigest()[:40]
 
 
-def user_avatar_path(user_profile: UserProfile) -> str:
-    # WARNING: If this method is changed, you may need to do a migration
-    # similar to zerver/migrations/0060_move_avatars_to_be_uid_based.py .
-    return user_avatar_path_from_ids(user_profile.id, user_profile.realm_id)
+def user_avatar_path(user_profile: UserProfile, future: bool = False) -> str:
+    # 'future' is if this is for the current avatar version, of the next one.
+    return user_avatar_base_path_from_ids(
+        user_profile.id, user_profile.avatar_version + (1 if future else 0), user_profile.realm_id
+    )
 
 
-def user_avatar_path_from_ids(user_profile_id: int, realm_id: int) -> str:
-    user_id_hash = user_avatar_hash(str(user_profile_id))
+def user_avatar_base_path_from_ids(user_profile_id: int, version: int, realm_id: int) -> str:
+    user_id_hash = user_avatar_hash(str(user_profile_id), str(version))
     return f"{realm_id}/{user_id_hash}"
 
 
diff --git a/zerver/lib/export.py b/zerver/lib/export.py
index be10b5b7d0..149a406c76 100644
--- a/zerver/lib/export.py
+++ b/zerver/lib/export.py
@@ -29,7 +29,7 @@ from typing_extensions import TypeAlias
 import zerver.lib.upload
 from analytics.models import RealmCount, StreamCount, UserCount
 from scripts.lib.zulip_tools import overwrite_symlink
-from zerver.lib.avatar_hash import user_avatar_path_from_ids
+from zerver.lib.avatar_hash import user_avatar_base_path_from_ids
 from zerver.lib.pysa import mark_sanitized
 from zerver.lib.upload.s3 import get_bucket
 from zerver.models import (
@@ -1543,8 +1543,10 @@ def export_uploads_and_avatars(
         )
 
         avatar_hash_values = set()
-        for user_id in user_ids:
-            avatar_path = user_avatar_path_from_ids(user_id, realm.id)
+        for avatar_user in users:
+            avatar_path = user_avatar_base_path_from_ids(
+                avatar_user.id, avatar_user.avatar_version, realm.id
+            )
             avatar_hash_values.add(avatar_path)
             avatar_hash_values.add(avatar_path + ".original")
 
@@ -1624,6 +1626,9 @@ def _get_exported_s3_record(
     else:
         raise Exception("Missing realm_id")
 
+    if "avatar_version" in record:
+        record["avatar_version"] = int(record["avatar_version"])
+
     return record
 
 
@@ -1782,7 +1787,7 @@ def export_avatars_from_local(
         if user.avatar_source == UserProfile.AVATAR_FROM_GRAVATAR:
             continue
 
-        avatar_path = user_avatar_path_from_ids(user.id, realm.id)
+        avatar_path = user_avatar_base_path_from_ids(user.id, user.avatar_version, realm.id)
         wildcard = os.path.join(local_dir, avatar_path + ".*")
 
         for local_path in glob.glob(wildcard):
@@ -1800,6 +1805,7 @@ def export_avatars_from_local(
                 realm_id=realm.id,
                 user_profile_id=user.id,
                 user_profile_email=user.email,
+                avatar_version=user.avatar_version,
                 s3_path=fn,
                 path=fn,
                 size=stat.st_size,
diff --git a/zerver/lib/import_realm.py b/zerver/lib/import_realm.py
index 41822633e9..0ed611f37d 100644
--- a/zerver/lib/import_realm.py
+++ b/zerver/lib/import_realm.py
@@ -21,7 +21,7 @@ from analytics.models import RealmCount, StreamCount, UserCount
 from zerver.actions.create_realm import set_default_for_realm_permission_group_settings
 from zerver.actions.realm_settings import do_change_realm_plan_type
 from zerver.actions.user_settings import do_change_avatar_fields
-from zerver.lib.avatar_hash import user_avatar_path_from_ids
+from zerver.lib.avatar_hash import user_avatar_base_path_from_ids
 from zerver.lib.bulk_create import bulk_set_users_or_streams_recipient_fields
 from zerver.lib.export import DATE_FIELDS, Field, Path, Record, TableData, TableName
 from zerver.lib.markdown import markdown_convert
@@ -797,7 +797,9 @@ def process_avatars(record: Dict[str, Any]) -> None:
         return None
     user_profile = get_user_profile_by_id(record["user_profile_id"])
     if settings.LOCAL_AVATARS_DIR is not None:
-        avatar_path = user_avatar_path_from_ids(user_profile.id, record["realm_id"])
+        avatar_path = user_avatar_base_path_from_ids(
+            user_profile.id, user_profile.avatar_version, record["realm_id"]
+        )
         medium_file_path = os.path.join(settings.LOCAL_AVATARS_DIR, avatar_path) + "-medium.png"
         if os.path.exists(medium_file_path):
             # We remove the image here primarily to deal with
@@ -864,7 +866,9 @@ def import_uploads(
         if processing_avatars:
             # For avatars, we need to rehash the user ID with the
             # new server's avatar salt
-            relative_path = user_avatar_path_from_ids(record["user_profile_id"], record["realm_id"])
+            relative_path = user_avatar_base_path_from_ids(
+                record["user_profile_id"], record["avatar_version"], record["realm_id"]
+            )
             if record["s3_path"].endswith(".original"):
                 relative_path += ".original"
             else:
diff --git a/zerver/lib/test_helpers.py b/zerver/lib/test_helpers.py
index 3681cfafc3..6f8797c683 100644
--- a/zerver/lib/test_helpers.py
+++ b/zerver/lib/test_helpers.py
@@ -246,7 +246,7 @@ def avatar_disk_path(
     avatar_disk_path = os.path.join(
         settings.LOCAL_AVATARS_DIR,
         avatar_url_path.split("/")[-2],
-        avatar_url_path.split("/")[-1].split("?")[0],
+        avatar_url_path.split("/")[-1],
     )
     if original:
         return avatar_disk_path.replace(".png", ".original")
diff --git a/zerver/lib/transfer.py b/zerver/lib/transfer.py
index b48ecf2d66..763d5c1737 100644
--- a/zerver/lib/transfer.py
+++ b/zerver/lib/transfer.py
@@ -30,7 +30,7 @@ def _transfer_avatar_to_s3(user: UserProfile) -> None:
     file_path = os.path.join(settings.LOCAL_AVATARS_DIR, avatar_path)
     try:
         with open(file_path + ".original", "rb") as f:
-            upload_avatar_image(f, user, backend=s3backend)
+            upload_avatar_image(f, user, backend=s3backend, future=False)
             logging.info("Uploaded avatar for %s in realm %s", user.id, user.realm.name)
     except FileNotFoundError:
         pass
@@ -66,7 +66,7 @@ def _transfer_message_files_to_s3(attachment: Attachment) -> None:
                 guessed_type,
                 attachment.owner,
                 f.read(),
-                settings.S3_UPLOADS_STORAGE_CLASS,
+                storage_class=settings.S3_UPLOADS_STORAGE_CLASS,
             )
             logging.info("Uploaded message file in path %s", file_path)
     except FileNotFoundError:  # nocoverage
diff --git a/zerver/lib/upload/__init__.py b/zerver/lib/upload/__init__.py
index 119ce917d7..b8ee8cd1d0 100644
--- a/zerver/lib/upload/__init__.py
+++ b/zerver/lib/upload/__init__.py
@@ -11,7 +11,7 @@ from django.conf import settings
 from django.core.files.uploadedfile import UploadedFile
 from django.utils.translation import gettext as _
 
-from zerver.lib.avatar_hash import user_avatar_path
+from zerver.lib.avatar_hash import user_avatar_base_path_from_ids, user_avatar_path
 from zerver.lib.exceptions import ErrorCode, JsonableError
 from zerver.lib.mime_types import guess_type
 from zerver.lib.outgoing_http import OutgoingSession
@@ -208,6 +208,7 @@ def write_avatar_images(
     *,
     content_type: Optional[str],
     backend: Optional[ZulipUploadBackend] = None,
+    future: bool = True,
 ) -> None:
     if backend is None:
         backend = upload_backend
@@ -216,6 +217,7 @@ def write_avatar_images(
         user_profile=user_profile,
         image_data=image_data,
         content_type=content_type,
+        future=future,
     )
 
     backend.upload_single_avatar_image(
@@ -223,6 +225,7 @@ def write_avatar_images(
         user_profile=user_profile,
         image_data=resize_avatar(image_data),
         content_type="image/png",
+        future=future,
     )
 
     backend.upload_single_avatar_image(
@@ -230,6 +233,7 @@ def write_avatar_images(
         user_profile=user_profile,
         image_data=resize_avatar(image_data, MEDIUM_AVATAR_SIZE),
         content_type="image/png",
+        future=future,
     )
 
 
@@ -238,23 +242,31 @@ def upload_avatar_image(
     user_profile: UserProfile,
     content_type: Optional[str] = None,
     backend: Optional[ZulipUploadBackend] = None,
+    future: bool = True,
 ) -> None:
     if content_type is None:
         content_type = guess_type(user_file.name)[0]
-    file_path = user_avatar_path(user_profile)
+    file_path = user_avatar_path(user_profile, future=future)
 
     image_data = user_file.read()
     write_avatar_images(
-        file_path, user_profile, image_data, content_type=content_type, backend=backend
+        file_path,
+        user_profile,
+        image_data,
+        content_type=content_type,
+        backend=backend,
+        future=future,
     )
 
 
 def copy_avatar(source_profile: UserProfile, target_profile: UserProfile) -> None:
-    source_file_path = user_avatar_path(source_profile)
-    target_file_path = user_avatar_path(target_profile)
+    source_file_path = user_avatar_path(source_profile, future=False)
+    target_file_path = user_avatar_path(target_profile, future=True)
 
     image_data, content_type = upload_backend.get_avatar_contents(source_file_path)
-    write_avatar_images(target_file_path, target_profile, image_data, content_type=content_type)
+    write_avatar_images(
+        target_file_path, target_profile, image_data, content_type=content_type, future=True
+    )
 
 
 def ensure_avatar_image(user_profile: UserProfile, medium: bool = False) -> None:
@@ -282,11 +294,12 @@ def ensure_avatar_image(user_profile: UserProfile, medium: bool = False) -> None
         user_profile=user_profile,
         image_data=resized_avatar,
         content_type="image/png",
+        future=False,
     )
 
 
-def delete_avatar_image(user_profile: UserProfile) -> None:
-    path_id = user_avatar_path(user_profile)
+def delete_avatar_image(user_profile: UserProfile, avatar_version: int) -> None:
+    path_id = user_avatar_base_path_from_ids(user_profile.id, avatar_version, user_profile.realm_id)
     upload_backend.delete_avatar_image(path_id)
 
 
diff --git a/zerver/lib/upload/base.py b/zerver/lib/upload/base.py
index b679e9cb8f..b0b42a8795 100644
--- a/zerver/lib/upload/base.py
+++ b/zerver/lib/upload/base.py
@@ -78,6 +78,7 @@ class ZulipUploadBackend:
         user_profile: UserProfile,
         image_data: bytes,
         content_type: Optional[str],
+        future: bool = True,
     ) -> None:
         raise NotImplementedError
 
diff --git a/zerver/lib/upload/local.py b/zerver/lib/upload/local.py
index 44b4781fb1..028ffbe29b 100644
--- a/zerver/lib/upload/local.py
+++ b/zerver/lib/upload/local.py
@@ -125,6 +125,7 @@ class LocalUploadBackend(ZulipUploadBackend):
         user_profile: UserProfile,
         image_data: bytes,
         content_type: Optional[str],
+        future: bool = True,
     ) -> None:
         write_local_file("avatars", file_path, image_data)
 
diff --git a/zerver/lib/upload/s3.py b/zerver/lib/upload/s3.py
index 113c29c48c..e1986d4ccf 100644
--- a/zerver/lib/upload/s3.py
+++ b/zerver/lib/upload/s3.py
@@ -2,7 +2,7 @@ import logging
 import os
 import secrets
 from datetime import datetime
-from typing import IO, Any, BinaryIO, Callable, Iterator, List, Literal, Optional, Tuple
+from typing import IO, Any, BinaryIO, Callable, Dict, Iterator, List, Literal, Optional, Tuple
 from urllib.parse import urljoin, urlsplit, urlunsplit
 
 import boto3
@@ -66,6 +66,7 @@ def upload_image_to_s3(
     content_type: Optional[str],
     user_profile: UserProfile,
     contents: bytes,
+    *,
     storage_class: Literal[
         "GLACIER_IR",
         "INTELLIGENT_TIERING",
@@ -75,12 +76,15 @@ def upload_image_to_s3(
         "STANDARD_IA",
     ] = "STANDARD",
     cache_control: Optional[str] = None,
+    extra_metadata: Optional[Dict[str, str]] = None,
 ) -> None:
     key = bucket.Object(file_name)
     metadata = {
         "user_profile_id": str(user_profile.id),
         "realm_id": str(user_profile.realm_id),
     }
+    if extra_metadata is not None:
+        metadata.update(extra_metadata)
 
     extras = {}
     if content_type is None:
@@ -219,7 +223,7 @@ class S3UploadBackend(ZulipUploadBackend):
             content_type,
             user_profile,
             file_data,
-            settings.S3_UPLOADS_STORAGE_CLASS,
+            storage_class=settings.S3_UPLOADS_STORAGE_CLASS,
         )
 
     @override
@@ -251,15 +255,6 @@ class S3UploadBackend(ZulipUploadBackend):
                         item["LastModified"],
                     )
 
-    @override
-    def get_avatar_path(self, hash_key: str, medium: bool = False) -> str:
-        # BUG: The else case should be f"{hash_key}.png".
-        # See #12852 for details on this bug and how to migrate it.
-        if medium:
-            return f"{hash_key}-medium.png"
-        else:
-            return hash_key
-
     @override
     def get_avatar_url(self, hash_key: str, medium: bool = False) -> str:
         return self.get_public_upload_url(self.get_avatar_path(hash_key, medium))
@@ -279,13 +274,17 @@ class S3UploadBackend(ZulipUploadBackend):
         user_profile: UserProfile,
         image_data: bytes,
         content_type: Optional[str],
+        future: bool = True,
     ) -> None:
+        extra_metadata = {"avatar_version": str(user_profile.avatar_version + (1 if future else 0))}
         upload_image_to_s3(
             self.avatar_bucket,
             file_path,
             content_type,
             user_profile,
             image_data,
+            extra_metadata=extra_metadata,
+            cache_control="public, max-age=31536000, immutable",
         )
 
     @override
diff --git a/zerver/migrations/0544_copy_avatar_images.py b/zerver/migrations/0544_copy_avatar_images.py
new file mode 100644
index 0000000000..607a2d1dcd
--- /dev/null
+++ b/zerver/migrations/0544_copy_avatar_images.py
@@ -0,0 +1,172 @@
+import contextlib
+import hashlib
+import os
+from typing import Any, Optional, Union
+
+import boto3
+import pyvips
+from botocore.client import Config
+from django.conf import settings
+from django.db import migrations
+from django.db.backends.base.schema import BaseDatabaseSchemaEditor
+from django.db.migrations.state import StateApps
+from django.db.models import QuerySet
+
+IMAGE_BOMB_TOTAL_PIXELS = 90000000
+DEFAULT_AVATAR_SIZE = 100
+MEDIUM_AVATAR_SIZE = 500
+
+
+def resize_avatar(
+    image_data: Union[bytes, pyvips.Image],
+    size: int,
+) -> Optional[bytes]:
+    try:
+        source_image = pyvips.Image.new_from_buffer(image_data, "")
+        if source_image.width * source_image.height > IMAGE_BOMB_TOTAL_PIXELS:
+            return None
+
+        return pyvips.Image.thumbnail_buffer(
+            image_data,
+            size,
+            height=size,
+            crop=pyvips.Interesting.CENTRE,
+        ).write_to_buffer(".png")
+    except pyvips.Error:
+        return None
+
+
+def new_hash(user_profile: Any) -> str:
+    user_key = (
+        str(user_profile.id) + ":" + str(user_profile.avatar_version) + ":" + settings.AVATAR_SALT
+    )
+    return hashlib.sha256(user_key.encode()).hexdigest()[:40]
+
+
+def old_hash(user_profile: Any) -> str:
+    user_key = str(user_profile.id) + settings.AVATAR_SALT
+    return hashlib.sha1(user_key.encode()).hexdigest()
+
+
+def thumbnail_s3_avatars(users: QuerySet[Any]) -> None:
+    avatar_bucket = boto3.resource(
+        "s3",
+        aws_access_key_id=settings.S3_KEY,
+        aws_secret_access_key=settings.S3_SECRET_KEY,
+        region_name=settings.S3_REGION,
+        endpoint_url=settings.S3_ENDPOINT_URL,
+        config=Config(
+            signature_version=None,
+            s3={"addressing_style": settings.S3_ADDRESSING_STYLE},
+        ),
+    ).Bucket(settings.S3_AVATAR_BUCKET)
+    for user in users:
+        old_base = os.path.join(str(user.realm_id), old_hash(user))
+        new_base = os.path.join(str(user.realm_id), new_hash(user))
+
+        try:
+            old_data = avatar_bucket.Object(old_base + ".original").get()
+            metadata = old_data["Metadata"]
+            metadata["avatar_version"] = str(user.avatar_version)
+            original_bytes = old_data["Body"].read()
+        except Exception:
+            print(f"Failed to fetch {old_base}")
+
+        avatar_bucket.Object(new_base + ".original").copy_from(
+            CopySource=f"{settings.S3_AVATAR_BUCKET}/{old_base}.original",
+            MetadataDirective="REPLACE",
+            Metadata=metadata,
+            ContentDisposition=old_data["ContentDisposition"],
+            ContentType=old_data["ContentType"],
+            CacheControl="public, max-age=31536000, immutable",
+        )
+
+        small = resize_avatar(original_bytes, DEFAULT_AVATAR_SIZE)
+        if small is None:
+            print(f"Failed to resize {old_base}")
+            continue
+        avatar_bucket.Object(new_base + ".png").put(
+            Metadata=metadata,
+            ContentDisposition=old_data["ContentDisposition"],
+            ContentType=old_data["ContentType"],
+            CacheControl="public, max-age=31536000, immutable",
+            Body=small,
+        )
+        medium = resize_avatar(original_bytes, MEDIUM_AVATAR_SIZE)
+        if medium is None:
+            print(f"Failed to medium resize {old_base}")
+            continue
+        avatar_bucket.Object(new_base + "-medium.png").put(
+            Metadata=metadata,
+            ContentDisposition=old_data["ContentDisposition"],
+            ContentType=old_data["ContentType"],
+            CacheControl="public, max-age=31536000, immutable",
+            Body=medium,
+        )
+
+
+def thumbnail_local_avatars(users: QuerySet[Any]) -> None:
+    total_processed = 0
+    assert settings.LOCAL_AVATARS_DIR is not None
+    for user in users:
+        old_base = os.path.join(settings.LOCAL_AVATARS_DIR, str(user.realm_id), old_hash(user))
+        new_base = os.path.join(settings.LOCAL_AVATARS_DIR, str(user.realm_id), new_hash(user))
+
+        if os.path.exists(new_base + "-medium.png"):
+            # This user's avatar has already been migrated.
+            continue
+
+        with contextlib.suppress(Exception):
+            # Remove the hard link, if present from a previous failed run.
+            os.remove(new_base + ".original")
+
+        # We hardlink, rather than copying, so we don't take any extra space.
+        try:
+            os.link(old_base + ".original", new_base + ".original")
+            with open(old_base + ".original", "rb") as f:
+                original_bytes = f.read()
+        except Exception:
+            print(f"Failed to read {old_base}.original")
+            raise
+
+        small = resize_avatar(original_bytes, DEFAULT_AVATAR_SIZE)
+        if small is None:
+            print(f"Failed to resize {old_base}")
+            continue
+        with open(new_base + ".png", "wb") as f:
+            f.write(small)
+
+        medium = resize_avatar(original_bytes, MEDIUM_AVATAR_SIZE)
+        if medium is None:
+            print(f"Failed to medium resize {old_base}")
+            continue
+        with open(new_base + "-medium.png", "wb") as f:
+            f.write(medium)
+
+        total_processed += 1
+        if total_processed % 100 == 0:
+            print(f"Processed {total_processed}/{len(users)} user avatars")
+
+
+def thumbnail_avatars(apps: StateApps, schema_editor: BaseDatabaseSchemaEditor) -> None:
+    UserProfile = apps.get_model("zerver", "UserProfile")
+    users = (
+        UserProfile.objects.filter(avatar_source="U")
+        .only("id", "realm_id", "avatar_version")
+        .order_by("id")
+    )
+    if settings.LOCAL_AVATARS_DIR is not None:
+        thumbnail_local_avatars(users)
+    else:
+        thumbnail_s3_avatars(users)
+
+
+class Migration(migrations.Migration):
+    atomic = False
+    elidable = True
+
+    dependencies = [
+        ("zerver", "0543_preregistrationuser_notify_referrer_on_join"),
+    ]
+
+    operations = [migrations.RunPython(thumbnail_avatars, elidable=True)]
diff --git a/zerver/tests/test_import_export.py b/zerver/tests/test_import_export.py
index 6e77f882ee..0127927edd 100644
--- a/zerver/tests/test_import_export.py
+++ b/zerver/tests/test_import_export.py
@@ -175,7 +175,7 @@ class ExportFile(ZulipTestCase):
         )
 
         with get_test_image_file("img.png") as img_file:
-            upload_avatar_image(img_file, user_profile)
+            upload_avatar_image(img_file, user_profile, future=False)
 
         user_profile.avatar_source = "U"
         user_profile.save()
diff --git a/zerver/tests/test_transfer.py b/zerver/tests/test_transfer.py
index f91ad22b51..4e957795d8 100644
--- a/zerver/tests/test_transfer.py
+++ b/zerver/tests/test_transfer.py
@@ -49,7 +49,7 @@ class TransferUploadsToS3Test(ZulipTestCase):
             transfer_avatars_to_s3(1)
 
         path_id = user_avatar_path(user)
-        image_key = bucket.Object(path_id)
+        image_key = bucket.Object(path_id + ".png")
         original_image_key = bucket.Object(path_id + ".original")
         medium_image_key = bucket.Object(path_id + "-medium.png")
 
diff --git a/zerver/tests/test_upload.py b/zerver/tests/test_upload.py
index 1ca582e41f..0e0b905ed2 100644
--- a/zerver/tests/test_upload.py
+++ b/zerver/tests/test_upload.py
@@ -948,7 +948,7 @@ class AvatarTest(UploadSerializeMixin, ZulipTestCase):
 
         self.assertEqual(
             url,
-            "/user_avatars/5/fc2b9f1a81f4508a4df2d95451a2a77e0524ca0e-medium.png?version=2",
+            "/user_avatars/5/ff062b0fee41738b38c4312bb33bdf3fe2aad463-medium.png",
         )
 
         url = get_avatar_field(
@@ -998,7 +998,7 @@ class AvatarTest(UploadSerializeMixin, ZulipTestCase):
         with self.settings(S3_AVATAR_BUCKET="bucket"):
             backend = S3UploadBackend()
             self.assertEqual(
-                backend.get_avatar_url("hash", False), "https://bucket.s3.amazonaws.com/hash"
+                backend.get_avatar_url("hash", False), "https://bucket.s3.amazonaws.com/hash.png"
             )
             self.assertEqual(
                 backend.get_avatar_url("hash", True),
@@ -1104,11 +1104,11 @@ class AvatarTest(UploadSerializeMixin, ZulipTestCase):
         cordelia.save()
         response = self.client_get("/avatar/cordelia@zulip.com", {"foo": "bar"})
         redirect_url = response["Location"]
-        self.assertTrue(redirect_url.endswith(str(avatar_url(cordelia)) + "&foo=bar"))
+        self.assertTrue(redirect_url.endswith(str(avatar_url(cordelia)) + "?foo=bar"))
 
         response = self.client_get(f"/avatar/{cordelia.id}", {"foo": "bar"})
         redirect_url = response["Location"]
-        self.assertTrue(redirect_url.endswith(str(avatar_url(cordelia)) + "&foo=bar"))
+        self.assertTrue(redirect_url.endswith(str(avatar_url(cordelia)) + "?foo=bar"))
 
         response = self.client_get("/avatar/")
         self.assertEqual(response.status_code, 404)
@@ -1119,11 +1119,11 @@ class AvatarTest(UploadSerializeMixin, ZulipTestCase):
             # Test /avatar/<email_or_id> endpoint with HTTP basic auth.
             response = self.api_get(hamlet, "/avatar/cordelia@zulip.com", {"foo": "bar"})
             redirect_url = response["Location"]
-            self.assertTrue(redirect_url.endswith(str(avatar_url(cordelia)) + "&foo=bar"))
+            self.assertTrue(redirect_url.endswith(str(avatar_url(cordelia)) + "?foo=bar"))
 
             response = self.api_get(hamlet, f"/avatar/{cordelia.id}", {"foo": "bar"})
             redirect_url = response["Location"]
-            self.assertTrue(redirect_url.endswith(str(avatar_url(cordelia)) + "&foo=bar"))
+            self.assertTrue(redirect_url.endswith(str(avatar_url(cordelia)) + "?foo=bar"))
 
             # Test cross_realm_bot avatar access using email.
             response = self.api_get(hamlet, "/avatar/welcome-bot@zulip.com", {"foo": "bar"})
@@ -1179,11 +1179,11 @@ class AvatarTest(UploadSerializeMixin, ZulipTestCase):
         cordelia.save()
         response = self.client_get("/avatar/cordelia@zulip.com/medium", {"foo": "bar"})
         redirect_url = response["Location"]
-        self.assertTrue(redirect_url.endswith(str(avatar_url(cordelia, True)) + "&foo=bar"))
+        self.assertTrue(redirect_url.endswith(str(avatar_url(cordelia, True)) + "?foo=bar"))
 
         response = self.client_get(f"/avatar/{cordelia.id}/medium", {"foo": "bar"})
         redirect_url = response["Location"]
-        self.assertTrue(redirect_url.endswith(str(avatar_url(cordelia, True)) + "&foo=bar"))
+        self.assertTrue(redirect_url.endswith(str(avatar_url(cordelia, True)) + "?foo=bar"))
 
         self.logout()
 
@@ -1191,11 +1191,11 @@ class AvatarTest(UploadSerializeMixin, ZulipTestCase):
             # Test /avatar/<email_or_id>/medium endpoint with HTTP basic auth.
             response = self.api_get(hamlet, "/avatar/cordelia@zulip.com/medium", {"foo": "bar"})
             redirect_url = response["Location"]
-            self.assertTrue(redirect_url.endswith(str(avatar_url(cordelia, True)) + "&foo=bar"))
+            self.assertTrue(redirect_url.endswith(str(avatar_url(cordelia, True)) + "?foo=bar"))
 
             response = self.api_get(hamlet, f"/avatar/{cordelia.id}/medium", {"foo": "bar"})
             redirect_url = response["Location"]
-            self.assertTrue(redirect_url.endswith(str(avatar_url(cordelia, True)) + "&foo=bar"))
+            self.assertTrue(redirect_url.endswith(str(avatar_url(cordelia, True)) + "?foo=bar"))
 
             # Without spectators enabled, no unauthenticated access.
             response = self.client_get("/avatar/cordelia@zulip.com/medium", {"foo": "bar"})
diff --git a/zerver/tests/test_upload_s3.py b/zerver/tests/test_upload_s3.py
index 186c57dc08..d1608a8b20 100644
--- a/zerver/tests/test_upload_s3.py
+++ b/zerver/tests/test_upload_s3.py
@@ -283,7 +283,7 @@ class S3Test(ZulipTestCase):
         medium_path_id = path_id + "-medium.png"
 
         with get_test_image_file("img.png") as image_file:
-            zerver.lib.upload.upload_avatar_image(image_file, user_profile)
+            zerver.lib.upload.upload_avatar_image(image_file, user_profile, future=False)
         test_image_data = read_test_image_file("img.png")
         test_medium_image_data = resize_avatar(test_image_data, MEDIUM_AVATAR_SIZE)
 
@@ -319,9 +319,9 @@ class S3Test(ZulipTestCase):
         target_path_id = user_avatar_path(target_user_profile)
         self.assertNotEqual(source_path_id, target_path_id)
 
-        source_image_key = bucket.Object(source_path_id)
-        target_image_key = bucket.Object(target_path_id)
-        self.assertEqual(target_image_key.key, target_path_id)
+        source_image_key = bucket.Object(source_path_id + ".png")
+        target_image_key = bucket.Object(target_path_id + ".png")
+        self.assertEqual(target_image_key.key, target_path_id + ".png")
         self.assertEqual(source_image_key.content_type, target_image_key.content_type)
         source_image_data = source_image_key.get()["Body"].read()
         target_image_data = target_image_key.get()["Body"].read()
@@ -354,13 +354,12 @@ class S3Test(ZulipTestCase):
 
         user_profile = self.example_user("hamlet")
         base_file_path = user_avatar_path(user_profile)
-        # Bug: This should have + ".png", but the implementation is wrong.
-        file_path = base_file_path
+        file_path = base_file_path + ".png"
         original_file_path = base_file_path + ".original"
         medium_file_path = base_file_path + "-medium.png"
 
         with get_test_image_file("img.png") as image_file:
-            zerver.lib.upload.upload_avatar_image(image_file, user_profile)
+            zerver.lib.upload.upload_avatar_image(image_file, user_profile, future=False)
 
         key = bucket.Object(original_file_path)
         image_data = key.get()["Body"].read()
@@ -385,9 +384,10 @@ class S3Test(ZulipTestCase):
 
         user = self.example_user("hamlet")
 
-        avatar_path_id = user_avatar_path(user)
-        avatar_original_image_path_id = avatar_path_id + ".original"
-        avatar_medium_path_id = avatar_path_id + "-medium.png"
+        avatar_base_path = user_avatar_path(user)
+        avatar_path_id = avatar_base_path + ".png"
+        avatar_original_image_path_id = avatar_base_path + ".original"
+        avatar_medium_path_id = avatar_base_path + "-medium.png"
 
         self.assertEqual(user.avatar_source, UserProfile.AVATAR_FROM_USER)
         self.assertIsNotNone(bucket.Object(avatar_path_id))
diff --git a/zerver/tests/test_users.py b/zerver/tests/test_users.py
index de3d690006..21e3db828a 100644
--- a/zerver/tests/test_users.py
+++ b/zerver/tests/test_users.py
@@ -1310,7 +1310,7 @@ class UserProfileTest(ZulipTestCase):
 
         # Upload cordelia's avatar
         with get_test_image_file("img.png") as image_file:
-            upload_avatar_image(image_file, cordelia)
+            upload_avatar_image(image_file, cordelia, future=False)
 
         OnboardingStep.objects.filter(user=cordelia).delete()
         OnboardingStep.objects.filter(user=iago).delete()
diff --git a/zerver/views/upload.py b/zerver/views/upload.py
index b8393ab1b2..8815f0ae5f 100644
--- a/zerver/views/upload.py
+++ b/zerver/views/upload.py
@@ -286,8 +286,6 @@ def serve_local_avatar_unauthed(request: HttpRequest, path: str) -> HttpResponse
         # backend; however, there is no reason to not serve the
         # redirect to S3 where the content lives.
         url = get_public_upload_root_url() + path
-        if request.GET.urlencode():
-            url += "?" + request.GET.urlencode()
         return redirect(url, permanent=True)
 
     local_path = os.path.join(settings.LOCAL_AVATARS_DIR, path)
@@ -300,10 +298,7 @@ def serve_local_avatar_unauthed(request: HttpRequest, path: str) -> HttpResponse
     else:
         response = internal_nginx_redirect(quote(f"/internal/local/user_avatars/{path}"))
 
-    # We do _not_ mark the contents as immutable for caching purposes,
-    # since the path for avatar images is hashed only by their user-id
-    # and a salt, and as such are reused when a user's avatar is
-    # updated.
+    patch_cache_control(response, max_age=31536000, public=True, immutable=True)
     return response
 
 
diff --git a/zerver/views/users.py b/zerver/views/users.py
index 749b408a82..bdbcab3e39 100644
--- a/zerver/views/users.py
+++ b/zerver/views/users.py
@@ -554,7 +554,7 @@ def add_bot_backend(
         [user_file] = request.FILES.values()
         assert isinstance(user_file, UploadedFile)
         assert user_file.size is not None
-        upload_avatar_image(user_file, bot_profile)
+        upload_avatar_image(user_file, bot_profile, future=False)
 
     if bot_type in (UserProfile.OUTGOING_WEBHOOK_BOT, UserProfile.EMBEDDED_BOT):
         assert isinstance(service_name, str)