diff --git a/README.md b/README.md
index d2b57f91..99f82bbc 100644
--- a/README.md
+++ b/README.md
@@ -629,6 +629,22 @@ If this an optional section that specifies keyring configuration parameters. It
   - 'h' (e.g. "42h") means hours ('42 * 60 * 60' seconds)
   - 'd' (e.g. "42d") means days ('42 * 60 * 60 *24' seconds)
 
+##### Storage consistency checks
+
+When the Binlog Server process is not properly shut down (killed, crashed, power loss), the storage may be left in an inconsistent state. Every time the storage is opened, the following hierarchy of sources of truth is used to check it:
+1. `binlog.index` defines the set of binlog files in the storage. Only objects that are known to be left after an interrupted operation of the Binlog Server are treated as garbage:
+   - the binlog data file (and its metadata file) that follows the last binlog file listed in `binlog.index` - left after an interrupted binlog rotation;
+   - binlog data / metadata files with the same base name as the first binlog file listed in `binlog.index` and a smaller sequence number - left after an interrupted purge;
+   - if `binlog.index` does not exist, a single binlog data file (and its metadata file) - left after an interrupted creation of the very first binlog file;
+   - temporary objects (`*.tmp`) left after interrupted writes.
+
+   Any other object not listed in `binlog.index` (including objects not created by the Binlog Server) makes the storage inconsistent.
+2. The binlog metadata file (`<binlog name>.json`) defines the size of the binlog data file. If the most recent binlog data file is bigger than recorded in its metadata, it is truncated to the recorded size.
+
+In the `fetch`, `pull` and `purge_binlogs` modes, the storage is fixed accordingly (garbage objects are removed after the whole storage has been checked) and the operation continues. The operation fails if the storage cannot be fixed automatically: a binlog file listed in `binlog.index` has no data file or no valid metadata file, a binlog data file is smaller than recorded in its metadata, the size mismatch is found in a binlog file that is not the most recent one, the storage contains an object not listed in `binlog.index` that is not garbage, or `binlog.index` is missing while the storage contains anything but a single binlog file. Nothing is removed from a storage that cannot be fixed. Please notice that these fixes are not safe to perform while another instance is fetching / pulling data to the same storage (see the note about the `purge_binlogs` mode above).
+
+In the `list`, `search_by_timestamp` and `search_by_gtid_set` modes, the storage is never modified, and any inconsistency found is ignored: these modes typically run as a second, read-only process just to inspect the storage, so the JSON response reflects whatever is on disk at the time, as-is (if `binlog.index` is missing while the storage contains anything but a single binlog file, the storage is considered empty).
+
 ##### Storage URI format
 
 - When `<storage.backend>` is set to `file`, `<storage.uri>` must be `file://...`.
diff --git a/mtr/binlog_streaming/include/assert_storage_object_absent.inc b/mtr/binlog_streaming/include/assert_storage_object_absent.inc
new file mode 100644
index 00000000..48effb5c
--- /dev/null
+++ b/mtr/binlog_streaming/include/assert_storage_object_absent.inc
@@ -0,0 +1,39 @@
+#
+# Asserts that an object does not exist in the backend storage.
+#
+# Usage:
+#   --let $storage_backend = file
+#   --let $storage_object = $MYSQL_TMP_DIR/first
+#   --source assert_storage_object_absent.inc
+#
+# or
+#   --let $aws_cli = AWS_ACCESS_KEY_ID=... AWS_SECRET_ACCESS_KEY=... aws
+#   --let $storage_backend = s3
+#   --let $aws_s3_bucket = my-bucket
+#   --let $storage_object = /vault/first
+#   --source assert_storage_object_absent.inc
+#
+# $storage_backend - determines storage backend type (either 'file' or 's3')
+# $aws_cli         - path to AWS command line interface (cli) tools with AWS_ACCESS_KEY_ID /
+#                    AWS_SECRET_ACCESS_KEY environment variables set appropriately
+#                    (needed only if $storage_backend is 's3')
+# $aws_s3_bucket   - AWS S3 bucket name (needed only if $storage_backend is 's3')
+# $storage_object  - if $storage_backend is 'file', a path to the file on a
+#                    local filesystem; if $storage_backend is 's3', a path to
+#                    the object on AWS S3
+#
+
+if ($storage_backend == file)
+{
+  --error 1
+  --file_exists $storage_object
+}
+if ($storage_backend == s3)
+{
+  # 'aws s3 ls' cannot be used here as it matches objects by prefix;
+  # 'aws s3api head-object' fails with exit code 254 (AWS CLI v2) or
+  # 255 (AWS CLI v1) when the object does not exist
+  --let $assert_storage_object_absent_key = `SELECT TRIM(LEADING '/' FROM '$storage_object')`
+  --error 254,255
+  --exec $aws_cli s3api head-object --bucket $aws_s3_bucket --key $assert_storage_object_absent_key > /dev/null 2>&1
+}
diff --git a/mtr/binlog_streaming/include/download_storage_object.inc b/mtr/binlog_streaming/include/download_storage_object.inc
new file mode 100644
index 00000000..f6942b36
--- /dev/null
+++ b/mtr/binlog_streaming/include/download_storage_object.inc
@@ -0,0 +1,37 @@
+#
+# Copies an object from the backend storage to a file on a local filesystem.
+#
+# Usage:
+#   --let $storage_backend = file
+#   --let $storage_object = $MYSQL_TMP_DIR/first
+#   --let $local_file = $MYSQL_TMP_DIR/second
+#   --source download_storage_object.inc
+#
+# or
+#   --let $aws_cli = AWS_ACCESS_KEY_ID=... AWS_SECRET_ACCESS_KEY=... aws
+#   --let $storage_backend = s3
+#   --let $aws_s3_bucket = my-bucket
+#   --let $storage_object = /vault/first
+#   --let $local_file = $MYSQL_TMP_DIR/second
+#   --source download_storage_object.inc
+#
+# $storage_backend - determines storage backend type (either 'file' or 's3')
+# $aws_cli         - path to AWS command line interface (cli) tools with AWS_ACCESS_KEY_ID /
+#                    AWS_SECRET_ACCESS_KEY environment variables set appropriately
+#                    (needed only if $storage_backend is 's3')
+# $aws_s3_bucket   - AWS S3 bucket name (needed only if $storage_backend is 's3')
+# $storage_object  - if $storage_backend is 'file', a path to the source file on
+#                    a local filesystem; if $storage_backend is 's3', a path to
+#                    the source object on AWS S3
+# $local_file      - a path to the destination file on a local filesystem
+#                    (must not exist)
+#
+
+if ($storage_backend == file)
+{
+  --copy_file $storage_object $local_file
+}
+if ($storage_backend == s3)
+{
+  --exec $aws_cli s3 cp s3://$aws_s3_bucket$storage_object $local_file > /dev/null
+}
diff --git a/mtr/binlog_streaming/include/remove_storage_object.inc b/mtr/binlog_streaming/include/remove_storage_object.inc
new file mode 100644
index 00000000..aa2a3aca
--- /dev/null
+++ b/mtr/binlog_streaming/include/remove_storage_object.inc
@@ -0,0 +1,33 @@
+#
+# Removes an object from the backend storage.
+#
+# Usage:
+#   --let $storage_backend = file
+#   --let $storage_object = $MYSQL_TMP_DIR/first
+#   --source remove_storage_object.inc
+#
+# or
+#   --let $aws_cli = AWS_ACCESS_KEY_ID=... AWS_SECRET_ACCESS_KEY=... aws
+#   --let $storage_backend = s3
+#   --let $aws_s3_bucket = my-bucket
+#   --let $storage_object = /vault/first
+#   --source remove_storage_object.inc
+#
+# $storage_backend - determines storage backend type (either 'file' or 's3')
+# $aws_cli         - path to AWS command line interface (cli) tools with AWS_ACCESS_KEY_ID /
+#                    AWS_SECRET_ACCESS_KEY environment variables set appropriately
+#                    (needed only if $storage_backend is 's3')
+# $aws_s3_bucket   - AWS S3 bucket name (needed only if $storage_backend is 's3')
+# $storage_object  - if $storage_backend is 'file', a path to the file on a
+#                    local filesystem; if $storage_backend is 's3', a path to
+#                    the object on AWS S3
+#
+
+if ($storage_backend == file)
+{
+  --remove_file $storage_object
+}
+if ($storage_backend == s3)
+{
+  --exec $aws_cli s3 rm s3://$aws_s3_bucket$storage_object > /dev/null
+}
diff --git a/mtr/binlog_streaming/include/upload_storage_object.inc b/mtr/binlog_streaming/include/upload_storage_object.inc
new file mode 100644
index 00000000..a4be609b
--- /dev/null
+++ b/mtr/binlog_streaming/include/upload_storage_object.inc
@@ -0,0 +1,41 @@
+#
+# Copies a file from a local filesystem to the backend storage (overwriting
+# the destination object if it already exists).
+#
+# Usage:
+#   --let $storage_backend = file
+#   --let $local_file = $MYSQL_TMP_DIR/first
+#   --let $storage_object = $MYSQL_TMP_DIR/second
+#   --source upload_storage_object.inc
+#
+# or
+#   --let $aws_cli = AWS_ACCESS_KEY_ID=... AWS_SECRET_ACCESS_KEY=... aws
+#   --let $storage_backend = s3
+#   --let $local_file = $MYSQL_TMP_DIR/first
+#   --let $aws_s3_bucket = my-bucket
+#   --let $storage_object = /vault/second
+#   --source upload_storage_object.inc
+#
+# $storage_backend - determines storage backend type (either 'file' or 's3')
+# $aws_cli         - path to AWS command line interface (cli) tools with AWS_ACCESS_KEY_ID /
+#                    AWS_SECRET_ACCESS_KEY environment variables set appropriately
+#                    (needed only if $storage_backend is 's3')
+# $local_file      - a path to the source file on a local filesystem
+# $aws_s3_bucket   - AWS S3 bucket name (needed only if $storage_backend is 's3')
+# $storage_object  - if $storage_backend is 'file', a path to the destination
+#                    file on a local filesystem; if $storage_backend is 's3', a
+#                    path to the destination object on AWS S3
+#
+
+if ($storage_backend == file)
+{
+  # '--copy_file' refuses to overwrite an existing file, hence removing the
+  # destination first (ignoring the case when it does not exist yet)
+  --error 0,1
+  --remove_file $storage_object
+  --copy_file $local_file $storage_object
+}
+if ($storage_backend == s3)
+{
+  --exec $aws_cli s3 cp $local_file s3://$aws_s3_bucket$storage_object > /dev/null
+}
diff --git a/mtr/binlog_streaming/r/storage_reconciliation.result b/mtr/binlog_streaming/r/storage_reconciliation.result
new file mode 100644
index 00000000..bfa5470c
--- /dev/null
+++ b/mtr/binlog_streaming/r/storage_reconciliation.result
@@ -0,0 +1,171 @@
+*** Resetting replication at the very beginning of the test.
+
+*** Generating two binary logs with some data in each of them.
+CREATE TABLE t1(id INT UNSIGNED NOT NULL AUTO_INCREMENT, PRIMARY KEY(id)) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(DEFAULT);
+FLUSH BINARY LOGS;
+INSERT INTO t1 VALUES(DEFAULT);
+
+*** Generating a configuration file in JSON format for the Binlog
+*** Server utility.
+
+*** Determining binlog file directory from the server.
+
+*** Creating a temporary directory <BINSRV_STORAGE_PATH> for storing
+*** binlog files downloaded via the Binlog Server utility.
+
+*** Executing the Binlog Server utility to download all binlog data
+*** from the server.
+
+*** Saving a copy of the second binlog data and metadata files to be
+*** used for simulating garbage objects.
+
+*** 1. Binlog files left after an interrupted binlog rotation are
+***    removed; querying-only operations ignore them without
+***    modifying the storage.
+
+*** Simulating a binlog rotation interrupted after the new binlog
+*** metadata was written but before the binlog index was updated, and
+*** an interrupted binlog index update.
+
+*** Executing the Binlog Server utility in the 'list' mode and
+*** expecting it to ignore the garbage objects without modifying the
+*** storage.
+include/assert_grep.inc [List response is not affected]
+
+*** Executing the Binlog Server utility and expecting it to remove
+*** garbage objects.
+include/assert_grep.inc [Found orphaned binlog data file]
+include/assert_grep.inc [Found orphaned binlog metadata file]
+include/assert_grep.inc [Removed orphaned objects]
+include/assert_grep.inc [Removed temporary object]
+
+*** 2. The most recent binlog data file is truncated to the size from
+***    its metadata; querying-only operations ignore the size mismatch
+***    without modifying the storage.
+
+*** Appending extra bytes to the second binlog data file.
+
+*** Executing the Binlog Server utility in the 'list' mode and
+*** expecting it to ignore the size mismatch without modifying the
+*** storage.
+include/assert_grep.inc [List response is not affected]
+
+*** Executing the Binlog Server utility and expecting it to truncate
+*** the second binlog data file.
+include/assert_grep.inc [Truncated the second binlog data file]
+
+*** 3. The storage with a missing metadata file of a binlog listed in
+***    the binlog index, with a size mismatch in an older binlog file or
+***    with an object that was not created by the Binlog Server cannot
+***    be repaired, and garbage objects are not removed from it.
+
+*** Simulating an interrupted binlog rotation one more time.
+
+*** Removing the first binlog metadata file.
+
+*** Executing the Binlog Server utility and expecting it to fail.
+include/assert_grep.inc [Refused to open storage with missing binlog metadata]
+include/assert_grep.inc [Nothing removed from the storage with missing binlog metadata]
+
+*** Restoring the first binlog metadata file.
+
+*** Appending extra bytes to the first (not the most recent) binlog
+*** data file.
+
+*** Executing the Binlog Server utility and expecting it to fail.
+include/assert_grep.inc [Refused to open storage with size mismatch in an older binlog]
+include/assert_grep.inc [Nothing removed from the storage with size mismatch in an older binlog]
+
+*** Restoring the first binlog data file.
+
+*** Adding an object that was not created by the Binlog Server.
+
+*** Executing the Binlog Server utility and expecting it to fail.
+include/assert_grep.inc [Refused to open storage with a foreign object]
+include/assert_grep.inc [Nothing removed from the storage with a foreign object]
+
+*** Removing the foreign object and the garbage objects.
+
+*** 4. The storage with binlog files but without the binlog index cannot
+***    be repaired; querying-only operations just ignore it.
+
+*** Removing the binlog index.
+
+*** Executing the Binlog Server utility in the 'list' mode and
+*** expecting an empty response.
+include/assert_grep.inc [List response contains no binlog files]
+
+*** Executing the Binlog Server utility and expecting it to fail.
+include/assert_grep.inc [Refused to open storage without binlog index]
+include/assert_grep.inc [Nothing removed from the storage without binlog index]
+
+*** Restoring the binlog index.
+
+*** Flushing the second binary log and dropping the table.
+FLUSH BINARY LOGS;
+DROP TABLE t1;
+
+*** Executing the Binlog Server utility to receive the rest of the
+*** events.
+
+*** 5. The 'purge_binlogs' operation removes garbage objects left after
+***    an improper shutdown as well.
+
+*** Saving a copy of the first binlog data and metadata files to be
+*** used for simulating an interrupted binlog purge.
+
+*** Simulating an interrupted binlog rotation one more time (the third
+*** binlog file is already in the storage now).
+
+*** Executing the Binlog Server utility in the 'purge_binlogs' mode.
+include/assert_grep.inc [Purge response is not affected]
+
+*** 6. Binlog files left after an interrupted binlog purge are removed.
+
+*** Simulating a binlog purge interrupted after the binlog index was
+*** updated but before the purged binlog files were removed.
+
+*** Executing the Binlog Server utility and expecting it to remove
+*** garbage objects.
+include/assert_grep.inc [Found purged binlog data file]
+include/assert_grep.inc [Removed purged binlog files]
+
+*** 7. Querying-only operations do not crash even if the metadata of
+***    every binlog file listed in the binlog index is missing (PBS-79).
+
+*** Removing the metadata files of all remaining binlog files.
+
+*** Executing the Binlog Server utility in the 'list' mode and
+*** expecting it to exit cleanly with an empty response.
+include/assert_grep.inc [List response contains no binlog files]
+
+*** 8. Querying-only operations ignore, and streaming operations
+***    recover, a storage left after an interrupted creation of the
+***    first binlog file (no binlog index, at most one binlog's data
+***    and metadata).
+
+*** Removing the remaining binlog data files and the binlog index.
+
+*** Simulating an interrupted creation of the first binlog file after
+*** both its data and metadata were written.
+
+*** Executing the Binlog Server utility in the 'list' mode and
+*** expecting it to ignore the leftover objects without modifying the
+*** storage.
+include/assert_grep.inc [List response contains no binlog files]
+
+*** Removing the metadata file to simulate an interrupted creation of
+*** the first binlog file after only its data was written.
+
+*** Executing the Binlog Server utility and expecting it to recover by
+*** removing the leftover binlog data file.
+include/assert_grep.inc [Found orphaned first binlog data file]
+include/assert_grep.inc [No orphaned first binlog metadata file found]
+include/assert_grep.inc [Removed orphaned first binlog file]
+
+*** Removing the Binlog Server utility storage directory.
+
+*** Removing the Binlog Server utility log file.
+
+*** Removing the Binlog Server utility configuration file.
diff --git a/mtr/binlog_streaming/t/storage_reconciliation.combinations b/mtr/binlog_streaming/t/storage_reconciliation.combinations
new file mode 100644
index 00000000..b75a1769
--- /dev/null
+++ b/mtr/binlog_streaming/t/storage_reconciliation.combinations
@@ -0,0 +1,5 @@
+[plain]
+init-connect = SET @binsrv_encryption = 'off'
+
+[encrypted]
+init-connect = SET @binsrv_encryption = 'on'
diff --git a/mtr/binlog_streaming/t/storage_reconciliation.test b/mtr/binlog_streaming/t/storage_reconciliation.test
new file mode 100644
index 00000000..3558928d
--- /dev/null
+++ b/mtr/binlog_streaming/t/storage_reconciliation.test
@@ -0,0 +1,550 @@
+#
+# Checks how the storage is reconciled when opened, using 'binlog.index' as
+# the source of truth for the set of binlog files and binlog metadata files
+# as the source of truth for binlog data file sizes:
+# - binlog files left after an interrupted binlog rotation (the one that
+#   follows the last binlog file from the binlog index) or after an
+#   interrupted binlog purge (the ones that precede the first binlog file
+#   from the binlog index) are garbage that is removed;
+# - any other object not referenced in the binlog index (including objects
+#   not created by the Binlog Server) makes the storage unrepairable;
+# - nothing is removed from a storage that cannot be repaired;
+# - querying-only operations never modify the storage and ignore any
+#   inconsistency they find, leaving their JSON response intact;
+# - the 'purge_binlogs' operation removes garbage objects;
+# - the storage that cannot be repaired automatically is not opened by a
+#   streaming operation ('fetch' / 'pull').
+# In the 'encrypted' combination all binlog files are encrypted, which also
+# exercises truncating an encrypted binlog data file and resuming its
+# encryption from an arbitrary (not block-aligned) offset.
+#
+--source ../include/have_binsrv.inc
+
+--source ../include/v80_v84_compatibility_defines.inc
+
+# in case of --repeat=N, we need to start from a fresh binary log to make
+# this test deterministic
+--echo *** Resetting replication at the very beginning of the test.
+--disable_query_log
+eval $stmt_reset_binary_logs_and_gtids;
+--enable_query_log
+
+--echo
+--echo *** Generating two binary logs with some data in each of them.
+--let $first_binlog = query_get_value($stmt_show_binary_log_status, File, 1)
+CREATE TABLE t1(id INT UNSIGNED NOT NULL AUTO_INCREMENT, PRIMARY KEY(id)) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(DEFAULT);
+FLUSH BINARY LOGS;
+--let $second_binlog = query_get_value($stmt_show_binary_log_status, File, 1)
+INSERT INTO t1 VALUES(DEFAULT);
+
+--disable_query_log
+--let $third_binlog = `SELECT CONCAT(SUBSTRING('$second_binlog', 1, LENGTH('$second_binlog') - 6), LPAD(CAST(RIGHT('$second_binlog', 6) AS UNSIGNED) + 1, 6, '0'))`
+--let $fourth_binlog = `SELECT CONCAT(SUBSTRING('$second_binlog', 1, LENGTH('$second_binlog') - 6), LPAD(CAST(RIGHT('$second_binlog', 6) AS UNSIGNED) + 2, 6, '0'))`
+--enable_query_log
+
+# identifying backend storage type ('file' or 's3')
+--source ../include/identify_storage_backend.inc
+
+# identifying whether binlog files must be encrypted from the combination
+--let $extracted_init_connect_variable_name = binsrv_encryption
+--source ../include/extract_init_connect_variable_value.inc
+if ($extracted_init_connect_variable_value == on)
+{
+  # generating keyring data file silently ('generate_keyring_data_file.inc'
+  # cannot be used as it echoes, and the output of both combinations must be
+  # the same - confirmed by recording this test with it, which left a
+  # combination-only echo in the shared result file)
+  --disable_query_log
+  SET @keyring_keys_array = JSON_ARRAY(
+    JSON_OBJECT('id', 'alpha16', 'cipher', 'AES-128-ECB', 'data_hex', '000102030405060708090A0B0C0D0E0F')
+  );
+  --enable_query_log
+  --let $binsrv_keyring_data_file_path = $MYSQL_TMP_DIR/keyring_data.json
+  --let $write_var = `SELECT JSON_OBJECT('version', 1, 'keys', CAST(@keyring_keys_array AS JSON))`
+  --let $write_to_file = $binsrv_keyring_data_file_path
+  --let $include_silent = 1
+  --source include/write_var_to_file.inc
+  --let $include_silent =
+  --let $binsrv_encryption_kek_id = alpha16
+  --let $binsrv_encryption_cipher = AES-256-CTR
+}
+
+# creating data directory, configuration file, etc.
+--let $binsrv_connect_timeout = 20
+--let $binsrv_read_timeout = 60
+--let $binsrv_idle_time = 10
+--let $binsrv_verify_checksum = TRUE
+--let $binsrv_replication_mode = position
+--let $binsrv_checkpoint_size = 1
+--source ../include/set_up_binsrv_environment.inc
+
+--let $saved_data_file = $MYSQL_TMP_DIR/storage_reconciliation.data
+--let $saved_metadata_file = $MYSQL_TMP_DIR/storage_reconciliation.json
+--let $response_file = $MYSQL_TMP_DIR/storage_reconciliation.response
+
+--echo
+--echo *** Executing the Binlog Server utility to download all binlog data
+--echo *** from the server.
+--exec $BINSRV fetch $binsrv_config_file_path > /dev/null
+
+--echo
+--echo *** Saving a copy of the second binlog data and metadata files to be
+--echo *** used for simulating garbage objects.
+--let $storage_object = $binsrv_storage_path/$second_binlog
+--let $local_file = $saved_data_file
+--source ../include/download_storage_object.inc
+--let $storage_object = $binsrv_storage_path/$second_binlog.json
+--let $local_file = $saved_metadata_file
+--source ../include/download_storage_object.inc
+
+--echo
+--echo *** 1. Binlog files left after an interrupted binlog rotation are
+--echo ***    removed; querying-only operations ignore them without
+--echo ***    modifying the storage.
+--echo
+--echo *** Simulating a binlog rotation interrupted after the new binlog
+--echo *** metadata was written but before the binlog index was updated, and
+--echo *** an interrupted binlog index update.
+--let $local_file = $saved_data_file
+--let $storage_object = $binsrv_storage_path/$third_binlog
+--source ../include/upload_storage_object.inc
+--let $local_file = $saved_metadata_file
+--let $storage_object = $binsrv_storage_path/$third_binlog.json
+--source ../include/upload_storage_object.inc
+--let $storage_object = $binsrv_storage_path/binlog.index.tmp
+--source ../include/upload_storage_object.inc
+
+--echo
+--echo *** Executing the Binlog Server utility in the 'list' mode and
+--echo *** expecting it to ignore the garbage objects without modifying the
+--echo *** storage.
+--exec $BINSRV list $binsrv_config_file_path > $response_file 2> /dev/null
+--let $assert_file = $response_file
+--let $assert_count = 0
+--let $assert_text = List response is not affected
+--let $assert_select = improper shutdown
+--source include/assert_grep.inc
+--remove_file $response_file
+
+--echo
+--echo *** Executing the Binlog Server utility and expecting it to remove
+--echo *** garbage objects.
+--exec $BINSRV fetch $binsrv_config_file_path > /dev/null
+
+--let $assert_file = $binsrv_log_path
+--let $assert_count = 1
+--let $assert_text = Found orphaned binlog data file
+--let $assert_select = found binlog data file '$third_binlog' left after an interrupted binlog rotation
+--source include/assert_grep.inc
+--let $assert_text = Found orphaned binlog metadata file
+--let $assert_select = found binlog metadata file '$third_binlog.json' left after an interrupted binlog rotation
+--source include/assert_grep.inc
+--let $assert_text = Removed orphaned objects
+--let $assert_select = removed 2 storage object\(s\) left after improper shutdown
+--source include/assert_grep.inc
+--let $assert_text = Removed temporary object
+--let $assert_select = removed 1 temporary storage object\(s\) left after improper shutdown
+--source include/assert_grep.inc
+
+--let $storage_object = $binsrv_storage_path/$third_binlog
+--source ../include/assert_storage_object_absent.inc
+--let $storage_object = $binsrv_storage_path/$third_binlog.json
+--source ../include/assert_storage_object_absent.inc
+--let $storage_object = $binsrv_storage_path/binlog.index.tmp
+--source ../include/assert_storage_object_absent.inc
+
+
+--echo
+--echo *** 2. The most recent binlog data file is truncated to the size from
+--echo ***    its metadata; querying-only operations ignore the size mismatch
+--echo ***    without modifying the storage.
+--echo
+--echo *** Appending extra bytes to the second binlog data file.
+--let $storage_object = $binsrv_storage_path/$second_binlog
+--let $local_file = $MYSQL_TMP_DIR/storage_reconciliation.local
+--source ../include/download_storage_object.inc
+--exec echo "extra bytes" >> $local_file
+--source ../include/upload_storage_object.inc
+
+--echo
+--echo *** Executing the Binlog Server utility in the 'list' mode and
+--echo *** expecting it to ignore the size mismatch without modifying the
+--echo *** storage.
+--exec $BINSRV list $binsrv_config_file_path > $response_file 2> /dev/null
+--let $assert_file = $response_file
+--let $assert_count = 0
+--let $assert_text = List response is not affected
+--let $assert_select = improper shutdown
+--source include/assert_grep.inc
+--remove_file $response_file
+
+# confirming the oversized data file was not truncated by the 'list' call
+# above by comparing it against the local copy appended to earlier
+--source ../include/diff_with_storage_object.inc
+--remove_file $local_file
+
+--echo
+--echo *** Executing the Binlog Server utility and expecting it to truncate
+--echo *** the second binlog data file.
+--exec $BINSRV fetch $binsrv_config_file_path > /dev/null
+--let $assert_file = $binsrv_log_path
+--let $assert_count = 1
+--let $assert_text = Truncated the second binlog data file
+--let $assert_select = recovered binlog file '$second_binlog' by truncating it to the size from the metadata
+--source include/assert_grep.inc
+
+
+--echo
+--echo *** 3. The storage with a missing metadata file of a binlog listed in
+--echo ***    the binlog index, with a size mismatch in an older binlog file or
+--echo ***    with an object that was not created by the Binlog Server cannot
+--echo ***    be repaired, and garbage objects are not removed from it.
+--echo
+--echo *** Simulating an interrupted binlog rotation one more time.
+--let $local_file = $saved_data_file
+--let $storage_object = $binsrv_storage_path/$third_binlog
+--source ../include/upload_storage_object.inc
+--let $local_file = $saved_metadata_file
+--let $storage_object = $binsrv_storage_path/$third_binlog.json
+--source ../include/upload_storage_object.inc
+
+--echo
+--echo *** Removing the first binlog metadata file.
+--let $storage_object = $binsrv_storage_path/$first_binlog.json
+--let $local_file = $MYSQL_TMP_DIR/storage_reconciliation.local
+--source ../include/download_storage_object.inc
+--source ../include/remove_storage_object.inc
+
+--echo
+--echo *** Executing the Binlog Server utility and expecting it to fail.
+--error 1
+--exec $BINSRV fetch $binsrv_config_file_path > /dev/null
+--let $assert_file = $binsrv_log_path
+--let $assert_count = 1
+--let $assert_text = Refused to open storage with missing binlog metadata
+--let $assert_select = missing metadata for a binlog listed in the binlog index
+--source include/assert_grep.inc
+--let $assert_count = 0
+--let $assert_text = Nothing removed from the storage with missing binlog metadata
+--let $assert_select = removed [0-9]+ storage object
+--source include/assert_grep.inc
+# downloading (and immediately discarding) the garbage object only to
+# confirm nothing was removed from the unrepairable storage
+--let $local_file = $MYSQL_TMP_DIR/storage_reconciliation.check
+--let $storage_object = $binsrv_storage_path/$third_binlog
+--source ../include/download_storage_object.inc
+--remove_file $local_file
+--let $storage_object = $binsrv_storage_path/$first_binlog.json
+--let $local_file = $MYSQL_TMP_DIR/storage_reconciliation.local
+
+--echo
+--echo *** Restoring the first binlog metadata file.
+--source ../include/upload_storage_object.inc
+--remove_file $local_file
+
+--echo
+--echo *** Appending extra bytes to the first (not the most recent) binlog
+--echo *** data file.
+--let $storage_object = $binsrv_storage_path/$first_binlog
+--let $local_file = $MYSQL_TMP_DIR/storage_reconciliation.local
+--source ../include/download_storage_object.inc
+--let $saved_first_binlog_file = $MYSQL_TMP_DIR/storage_reconciliation.first
+--copy_file $local_file $saved_first_binlog_file
+--exec echo "extra bytes" >> $local_file
+--source ../include/upload_storage_object.inc
+--remove_file $local_file
+
+--echo
+--echo *** Executing the Binlog Server utility and expecting it to fail.
+--error 1
+--exec $BINSRV fetch $binsrv_config_file_path > /dev/null
+--let $assert_file = $binsrv_log_path
+--let $assert_count = 1
+--let $assert_text = Refused to open storage with size mismatch in an older binlog
+--let $assert_select = size from the binlog metadata does not match the actual binlog size
+--source include/assert_grep.inc
+--let $assert_count = 0
+--let $assert_text = Nothing removed from the storage with size mismatch in an older binlog
+--let $assert_select = removed [0-9]+ storage object
+--source include/assert_grep.inc
+
+--echo
+--echo *** Restoring the first binlog data file.
+--let $local_file = $saved_first_binlog_file
+--source ../include/upload_storage_object.inc
+--remove_file $saved_first_binlog_file
+
+--echo
+--echo *** Adding an object that was not created by the Binlog Server.
+--let $local_file = $saved_metadata_file
+--let $storage_object = $binsrv_storage_path/notes.txt
+--source ../include/upload_storage_object.inc
+
+--echo
+--echo *** Executing the Binlog Server utility and expecting it to fail.
+--error 1
+--exec $BINSRV fetch $binsrv_config_file_path > /dev/null
+--let $assert_file = $binsrv_log_path
+--let $assert_count = 1
+--let $assert_text = Refused to open storage with a foreign object
+--let $assert_select = storage contains an object that is not referenced in the binlog index
+--source include/assert_grep.inc
+--let $assert_count = 0
+--let $assert_text = Nothing removed from the storage with a foreign object
+--let $assert_select = removed [0-9]+ storage object
+--source include/assert_grep.inc
+# downloading (and immediately discarding) the foreign object and the
+# garbage object only to confirm nothing was removed from the unrepairable
+# storage
+--let $local_file = $MYSQL_TMP_DIR/storage_reconciliation.check
+--source ../include/download_storage_object.inc
+--remove_file $local_file
+--let $storage_object = $binsrv_storage_path/$third_binlog
+--source ../include/download_storage_object.inc
+--remove_file $local_file
+
+--echo
+--echo *** Removing the foreign object and the garbage objects.
+--let $storage_object = $binsrv_storage_path/notes.txt
+--source ../include/remove_storage_object.inc
+--let $storage_object = $binsrv_storage_path/$third_binlog
+--source ../include/remove_storage_object.inc
+--let $storage_object = $binsrv_storage_path/$third_binlog.json
+--source ../include/remove_storage_object.inc
+
+
+--echo
+--echo *** 4. The storage with binlog files but without the binlog index cannot
+--echo ***    be repaired; querying-only operations just ignore it.
+--echo
+--echo *** Removing the binlog index.
+--let $storage_object = $binsrv_storage_path/binlog.index
+--let $local_file = $MYSQL_TMP_DIR/storage_reconciliation.local
+--source ../include/download_storage_object.inc
+--source ../include/remove_storage_object.inc
+
+--echo
+--echo *** Executing the Binlog Server utility in the 'list' mode and
+--echo *** expecting an empty response.
+--exec $BINSRV list $binsrv_config_file_path > $response_file 2> /dev/null
+--let $assert_file = $response_file
+--let $assert_count = 0
+--let $assert_text = List response contains no binlog files
+--let $assert_select = "name"
+--source include/assert_grep.inc
+--remove_file $response_file
+
+--echo
+--echo *** Executing the Binlog Server utility and expecting it to fail.
+--error 1
+--exec $BINSRV fetch $binsrv_config_file_path > /dev/null
+--let $assert_file = $binsrv_log_path
+--let $assert_count = 1
+--let $assert_text = Refused to open storage without binlog index
+--let $assert_select = storage is not empty but does not contain binlog index
+--source include/assert_grep.inc
+--let $assert_count = 0
+--let $assert_text = Nothing removed from the storage without binlog index
+--let $assert_select = removed [0-9]+ storage object
+--source include/assert_grep.inc
+
+--echo
+--echo *** Restoring the binlog index.
+--source ../include/upload_storage_object.inc
+--remove_file $local_file
+
+
+--echo
+--echo *** Flushing the second binary log and dropping the table.
+FLUSH BINARY LOGS;
+DROP TABLE t1;
+
+--echo
+--echo *** Executing the Binlog Server utility to receive the rest of the
+--echo *** events.
+--exec $BINSRV fetch $binsrv_config_file_path > /dev/null
+
+
+--echo
+--echo *** 5. The 'purge_binlogs' operation removes garbage objects left after
+--echo ***    an improper shutdown as well.
+--echo
+--echo *** Saving a copy of the first binlog data and metadata files to be
+--echo *** used for simulating an interrupted binlog purge.
+--let $saved_first_binlog_file = $MYSQL_TMP_DIR/storage_reconciliation.first
+--let $saved_first_metadata_file = $MYSQL_TMP_DIR/storage_reconciliation.first.json
+--let $storage_object = $binsrv_storage_path/$first_binlog
+--let $local_file = $saved_first_binlog_file
+--source ../include/download_storage_object.inc
+--let $storage_object = $binsrv_storage_path/$first_binlog.json
+--let $local_file = $saved_first_metadata_file
+--source ../include/download_storage_object.inc
+
+--echo
+--echo *** Simulating an interrupted binlog rotation one more time (the third
+--echo *** binlog file is already in the storage now).
+--let $local_file = $saved_data_file
+--let $storage_object = $binsrv_storage_path/$fourth_binlog
+--source ../include/upload_storage_object.inc
+--let $local_file = $saved_metadata_file
+--let $storage_object = $binsrv_storage_path/$fourth_binlog.json
+--source ../include/upload_storage_object.inc
+
+--echo
+--echo *** Executing the Binlog Server utility in the 'purge_binlogs' mode.
+--exec $BINSRV purge_binlogs $binsrv_config_file_path $first_binlog > $response_file 2> /dev/null
+
+--let $assert_file = $response_file
+--let $assert_count = 0
+--let $assert_text = Purge response is not affected
+--let $assert_select = improper shutdown
+--source include/assert_grep.inc
+--remove_file $response_file
+
+--let $storage_object = $binsrv_storage_path/$fourth_binlog
+--source ../include/assert_storage_object_absent.inc
+--let $storage_object = $binsrv_storage_path/$fourth_binlog.json
+--source ../include/assert_storage_object_absent.inc
+--let $storage_object = $binsrv_storage_path/$first_binlog
+--source ../include/assert_storage_object_absent.inc
+
+
+--echo
+--echo *** 6. Binlog files left after an interrupted binlog purge are removed.
+--echo
+--echo *** Simulating a binlog purge interrupted after the binlog index was
+--echo *** updated but before the purged binlog files were removed.
+--let $local_file = $saved_first_binlog_file
+--let $storage_object = $binsrv_storage_path/$first_binlog
+--source ../include/upload_storage_object.inc
+--let $local_file = $saved_first_metadata_file
+--let $storage_object = $binsrv_storage_path/$first_binlog.json
+--source ../include/upload_storage_object.inc
+--remove_file $saved_first_binlog_file
+--remove_file $saved_first_metadata_file
+
+--echo
+--echo *** Executing the Binlog Server utility and expecting it to remove
+--echo *** garbage objects.
+--exec $BINSRV fetch $binsrv_config_file_path > /dev/null
+--let $assert_file = $binsrv_log_path
+--let $assert_count = 1
+--let $assert_text = Found purged binlog data file
+--let $assert_select = found binlog data file '$first_binlog' left after an interrupted binlog purge
+--source include/assert_grep.inc
+--let $assert_text = Removed purged binlog files
+--let $assert_select = removed 2 storage object\(s\) left after improper shutdown
+--source include/assert_grep.inc
+
+--let $storage_object = $binsrv_storage_path/$first_binlog
+--source ../include/assert_storage_object_absent.inc
+--let $storage_object = $binsrv_storage_path/$first_binlog.json
+--source ../include/assert_storage_object_absent.inc
+
+
+--echo
+--echo *** 7. Querying-only operations do not crash even if the metadata of
+--echo ***    every binlog file listed in the binlog index is missing (PBS-79).
+--echo
+--echo *** Removing the metadata files of all remaining binlog files.
+--let $storage_object = $binsrv_storage_path/$second_binlog.json
+--source ../include/remove_storage_object.inc
+--let $storage_object = $binsrv_storage_path/$third_binlog.json
+--source ../include/remove_storage_object.inc
+
+--echo
+--echo *** Executing the Binlog Server utility in the 'list' mode and
+--echo *** expecting it to exit cleanly with an empty response.
+--exec $BINSRV list $binsrv_config_file_path > $response_file 2> /dev/null
+--let $assert_file = $response_file
+--let $assert_count = 0
+--let $assert_text = List response contains no binlog files
+--let $assert_select = "name"
+--source include/assert_grep.inc
+--remove_file $response_file
+
+
+--echo
+--echo *** 8. Querying-only operations ignore, and streaming operations
+--echo ***    recover, a storage left after an interrupted creation of the
+--echo ***    first binlog file (no binlog index, at most one binlog's data
+--echo ***    and metadata).
+--echo
+--echo *** Removing the remaining binlog data files and the binlog index.
+--let $storage_object = $binsrv_storage_path/$second_binlog
+--source ../include/remove_storage_object.inc
+--let $storage_object = $binsrv_storage_path/$third_binlog
+--source ../include/remove_storage_object.inc
+--let $storage_object = $binsrv_storage_path/binlog.index
+--source ../include/remove_storage_object.inc
+
+--disable_query_log
+--let $fifth_binlog = `SELECT CONCAT(SUBSTRING('$second_binlog', 1, LENGTH('$second_binlog') - 6), LPAD(CAST(RIGHT('$second_binlog', 6) AS UNSIGNED) + 3, 6, '0'))`
+--enable_query_log
+
+--echo
+--echo *** Simulating an interrupted creation of the first binlog file after
+--echo *** both its data and metadata were written.
+--let $local_file = $saved_data_file
+--let $storage_object = $binsrv_storage_path/$fifth_binlog
+--source ../include/upload_storage_object.inc
+--let $local_file = $saved_metadata_file
+--let $storage_object = $binsrv_storage_path/$fifth_binlog.json
+--source ../include/upload_storage_object.inc
+
+--echo
+--echo *** Executing the Binlog Server utility in the 'list' mode and
+--echo *** expecting it to ignore the leftover objects without modifying the
+--echo *** storage.
+--exec $BINSRV list $binsrv_config_file_path > $response_file 2> /dev/null
+--let $assert_file = $response_file
+--let $assert_count = 0
+--let $assert_text = List response contains no binlog files
+--let $assert_select = "name"
+--source include/assert_grep.inc
+--remove_file $response_file
+
+--echo
+--echo *** Removing the metadata file to simulate an interrupted creation of
+--echo *** the first binlog file after only its data was written.
+--let $storage_object = $binsrv_storage_path/$fifth_binlog.json
+--source ../include/remove_storage_object.inc
+
+--echo
+--echo *** Executing the Binlog Server utility and expecting it to recover by
+--echo *** removing the leftover binlog data file.
+--exec $BINSRV fetch $binsrv_config_file_path > /dev/null
+--let $assert_file = $binsrv_log_path
+--let $assert_count = 1
+--let $assert_text = Found orphaned first binlog data file
+--let $assert_select = found binlog data file '$fifth_binlog' left after an interrupted creation of the first binlog file
+--source include/assert_grep.inc
+--let $assert_count = 0
+--let $assert_text = No orphaned first binlog metadata file found
+--let $assert_select = found binlog metadata file '$fifth_binlog.json' left after an interrupted creation of the first binlog file
+--source include/assert_grep.inc
+--let $assert_count = 1
+--let $assert_text = Removed orphaned first binlog file
+--let $assert_select = removed 1 storage object\(s\) left after improper shutdown
+--source include/assert_grep.inc
+
+--let $storage_object = $binsrv_storage_path/$fifth_binlog
+--source ../include/assert_storage_object_absent.inc
+
+# cleaning up
+--remove_file $saved_data_file
+--remove_file $saved_metadata_file
+--source ../include/tear_down_binsrv_environment.inc
+if ($binsrv_keyring_data_file_path != "")
+{
+  # not using 'delete_keyring_data_file.inc' - same combination-parity
+  # reason as above
+  --remove_file $binsrv_keyring_data_file_path
+  --disable_query_log
+  SET @keyring_keys_array = NULL;
+  --enable_query_log
+  --let $binsrv_keyring_data_file_path =
+  --let $binsrv_encryption_kek_id =
+  --let $binsrv_encryption_cipher =
+}
diff --git a/src/binsrv/storage_core.cpp b/src/binsrv/storage_core.cpp
index b4f559a8..01972bce 100644
--- a/src/binsrv/storage_core.cpp
+++ b/src/binsrv/storage_core.cpp
@@ -184,15 +184,27 @@ storage_core::storage_core(basic_logger_ptr logger, const main_config &config,
     return;
   }
 
+  // objects left after improper shutdown are collected here and removed
+  // only after the whole storage has been validated, so that nothing is
+  // removed from a storage that cannot be repaired automatically
+  storage_object_name_list garbage_object_names;
+
   const auto binlog_index_it{storage_objects.find(default_binlog_index_name)};
   if (binlog_index_it == std::cend(storage_objects)) {
     // as binlog index file is created after the very first binlog data file
     // and its metadata file are created, the concurrent query-only operation
     // may intrude exactly between these two steps, so for query-only operations
-    // we should not consider the absence of binlog index file as an error
+    // we should not consider the absence of binlog index file as an error;
+    // the querying_only mode never modifies the storage, so it must return
+    // here without even attempting to detect (and log) leftover objects
     if (construction_mode == storage_construction_mode_type::querying_only) {
       return;
     }
+    if (collect_interrupted_first_binlog_objects(storage_objects,
+                                                 garbage_object_names)) {
+      remove_garbage_objects(garbage_object_names);
+      return;
+    }
     util::exception_location().raise<std::logic_error>(
         "storage is not empty but does not contain binlog index");
   }
@@ -212,12 +224,19 @@ storage_core::storage_core(basic_logger_ptr logger, const main_config &config,
     }
   }
   load_binlog_index();
+  collect_interrupted_rotation_objects(
+      storage_objects, storage_metadata_objects, garbage_object_names);
+  collect_interrupted_purge_objects(storage_objects, storage_metadata_objects,
+                                    garbage_object_names);
   validate_binlog_index(storage_objects);
 
   load_and_validate_binlog_metadata_set(storage_objects,
                                         storage_metadata_objects);
-  assert(!binlog_records_.front().added_gtids.has_value() ||
+  assert(binlog_records_.empty() ||
+         !binlog_records_.front().added_gtids.has_value() ||
          purged_gtids_ == binlog_records_.front().added_gtids);
+
+  remove_garbage_objects(garbage_object_names);
 }
 
 storage_core::~storage_core() = default;
@@ -383,18 +402,17 @@ storage_core::purge_binlogs(const events::composite_binlog_name &target) {
   // tail-refusal guard above). 'save_binlog_index' goes through the
   // backend's atomic-overwrite 'put_object', so from this point on
   // the purge is considered committed - any subsequent failure
-  // leaves the storage in an inconsistent state (leftover payload /
-  // metadata files no longer referenced by the index) that the
-  // constructor's existing validators will refuse to open on next
-  // startup.
+  // leaves leftover payload / metadata files no longer referenced by
+  // the index, which 'collect_interrupted_purge_objects()' picks up as
+  // garbage on next startup.
   save_binlog_index();
 
   // step 3: best-effort removal of the victim payload + metadata
   // objects; any failure here is intentionally swallowed - the index
   // has already been committed and reporting a "file could not be
   // removed" error to the caller would falsely suggest that the
-  // purge itself failed; the resulting leftovers will trip the
-  // constructor's validators on next startup.
+  // purge itself failed; the resulting leftovers will be removed as
+  // garbage on next startup.
   // We materialise the (metadata + payload) names for every victim
   // into a single batch and hand it to 'basic_storage_backend::
   // remove_objects', which runs the backend's durability barrier
@@ -415,8 +433,8 @@ storage_core::purge_binlogs(const events::composite_binlog_name &target) {
     // 'remove_objects' re-raises the first per-name failure (if any)
     // after running the durability barrier; we do not propagate it
     // to the caller because the index has already been committed
-    // and any leftover payload/metadata files will be picked up by
-    // the constructor's validators on the next startup. We just
+    // and any leftover payload/metadata files will be removed as
+    // garbage on the next startup. We just
     // capture the underlying message so the caller can surface it
     // under a 'warning' status in the JSON response.
     cleanup_warning_message = e.what();
@@ -604,6 +622,217 @@ void storage_core::remove_temporary_objects(
                       remove_objects.size());
 }
 
+[[nodiscard]] bool storage_core::collect_interrupted_first_binlog_objects(
+    storage_object_name_container &object_names,
+    storage_object_name_list &garbage_object_names) const {
+  // PBS does the following when it creates the very first binlog file in
+  // the storage:
+  // 1. Creates a new binlog data file
+  // 2. Creates a new binlog metadata file
+  // 3. Creates the binlog index with the new binlog data file entry
+
+  // Here we try to deal with the following scenarios:
+  // 1. If crash happens after (1), we will have only 1 binlog data file and
+  //    no binlog index.
+  // 2. If crash happens after (2), we will have 1 binlog data file, its
+  //    binlog metadata file and no binlog index.
+  // Anything else without the binlog index means that the binlog index was
+  // lost or that the storage contains objects not created by the Binlog
+  // Server - the storage cannot be repaired automatically in this case.
+
+  if (std::size(object_names) > 2U) {
+    return false;
+  }
+  const auto data_object_it{
+      std::ranges::find_if(object_names, [](const auto &object) {
+        const std::filesystem::path object_name{object.first};
+        return !object_name.has_extension() ||
+               object_name.extension() != binlog_metadata_extension;
+      })};
+  if (data_object_it == std::end(object_names)) {
+    return false;
+  }
+  auto data_object_name{data_object_it->first};
+  events::composite_binlog_name parsed_data_object_name;
+  try {
+    parsed_data_object_name =
+        events::composite_binlog_name::parse(data_object_name);
+  } catch (const std::exception &) {
+    return false;
+  }
+  auto metadata_object_name{
+      generate_binlog_metadata_name(parsed_data_object_name)};
+  const bool has_metadata_object{object_names.contains(metadata_object_name)};
+  if (std::size(object_names) == 2U && !has_metadata_object) {
+    return false;
+  }
+
+  // binlog metadata file goes first to the removal list - if the removal
+  // fails in the middle, we will end up in scenario (1) described above
+  if (has_metadata_object) {
+    logger_->log_format(log_severity::warning,
+                        "found binlog metadata file '{}' left after an "
+                        "interrupted creation of the first binlog file",
+                        metadata_object_name);
+    garbage_object_names.push_back(std::move(metadata_object_name));
+  }
+  logger_->log_format(log_severity::warning,
+                      "found binlog data file '{}' left after an interrupted "
+                      "creation of the first binlog file",
+                      data_object_name);
+  garbage_object_names.push_back(std::move(data_object_name));
+  object_names.clear();
+  return true;
+}
+
+void storage_core::collect_interrupted_rotation_objects(
+    storage_object_name_container &object_names,
+    storage_object_name_container &object_metadata_names,
+    storage_object_name_list &garbage_object_names) const {
+  // TODO: This automatic recovery procedure won't cover the case when the
+  //       crash happens during the binlog rotation with base name change
+  //       (e.g. 'binlog.000001' -> 'my_base.000001'). In this case, manual
+  //       user intervention will be required.
+
+  // PBS does the following on binlog rotation:
+  // 1. Creates a new binlog data file
+  // 2. Creates a new binlog metadata file
+  // 3. Updates the binlog index to include the new binlog data file entry
+
+  // Here we try to deal with the following scenarios:
+  // 1. If crash happens after (1), we will have only 1 orphaned binlog data
+  //    file - the one that follows the last binlog file from the binlog
+  //    index.
+  // 2. If crash happens after (2), we will have this orphaned binlog data
+  //    file and its orphaned binlog metadata file.
+  // 3. There should not be a situation when we have an orphaned binlog
+  //    metadata file without the corresponding orphaned binlog data file.
+
+  if (binlog_records_.empty()) {
+    return;
+  }
+
+  events::composite_binlog_name next_binlog_name;
+  try {
+    next_binlog_name = binlog_records_.back().name.next();
+  } catch (const std::exception &) {
+    // the last binlog file has the largest possible sequence number, so
+    // there cannot be a binlog file that follows it
+    return;
+  }
+
+  auto next_binlog_name_str{next_binlog_name.str()};
+  const auto object_it{object_names.find(next_binlog_name_str)};
+  if (object_it == std::end(object_names)) {
+    return;
+  }
+
+  // binlog metadata file goes first to the removal list - if the removal
+  // fails in the middle, we will end up in scenario (1), not (3)
+  auto next_binlog_metadata_name_str{
+      generate_binlog_metadata_name(next_binlog_name)};
+  const auto metadata_it{
+      object_metadata_names.find(next_binlog_metadata_name_str)};
+  if (metadata_it != std::end(object_metadata_names)) {
+    object_metadata_names.erase(metadata_it);
+    logger_->log_format(log_severity::warning,
+                        "found binlog metadata file '{}' left after an "
+                        "interrupted binlog rotation",
+                        next_binlog_metadata_name_str);
+    garbage_object_names.push_back(std::move(next_binlog_metadata_name_str));
+  }
+  object_names.erase(object_it);
+  logger_->log_format(log_severity::warning,
+                      "found binlog data file '{}' left after an interrupted "
+                      "binlog rotation",
+                      next_binlog_name_str);
+  garbage_object_names.push_back(std::move(next_binlog_name_str));
+}
+
+void storage_core::collect_interrupted_purge_objects(
+    storage_object_name_container &object_names,
+    storage_object_name_container &object_metadata_names,
+    storage_object_name_list &garbage_object_names) const {
+  // PBS does the following on binlog purge:
+  // 1. Updates the binlog index to exclude the purged binlog file entries
+  //    (purged binlog files always precede the remaining ones in the binlog
+  //    index and have the same base name as the first remaining one)
+  // 2. Removes the binlog metadata file and the binlog data file of each
+  //    purged binlog file
+
+  // Here we try to deal with the following scenario:
+  // 1. If crash happens after (1) or in the middle of (2), we will have
+  //    orphaned binlog metadata / data files of the purged binlog files -
+  //    files that have the same base name as the first binlog file from the
+  //    binlog index and a smaller sequence number.
+
+  if (binlog_records_.empty()) {
+    return;
+  }
+
+  const auto &first_binlog_name{binlog_records_.front().name};
+  const auto is_purged_binlog_name{
+      [&first_binlog_name](std::string_view binlog_name) {
+        try {
+          const auto parsed_binlog_name{
+              events::composite_binlog_name::parse(binlog_name)};
+          return parsed_binlog_name.get_base_name() ==
+                     first_binlog_name.get_base_name() &&
+                 parsed_binlog_name.get_sequence_number() <
+                     first_binlog_name.get_sequence_number();
+        } catch (const std::exception &) {
+          return false;
+        }
+      }};
+
+  // binlog metadata files go first to the removal list (in the same order
+  // as in purge step (2))
+  for (auto it{std::begin(object_metadata_names)};
+       it != std::end(object_metadata_names);) {
+    const auto binlog_name{std::filesystem::path{it->first}.stem().string()};
+    if (!is_purged_binlog_name(binlog_name)) {
+      ++it;
+      continue;
+    }
+    logger_->log_format(log_severity::warning,
+                        "found binlog metadata file '{}' left after an "
+                        "interrupted binlog purge",
+                        it->first);
+    garbage_object_names.push_back(it->first);
+    it = object_metadata_names.erase(it);
+  }
+  for (auto it{std::begin(object_names)}; it != std::end(object_names);) {
+    if (!is_purged_binlog_name(it->first)) {
+      ++it;
+      continue;
+    }
+    logger_->log_format(log_severity::warning,
+                        "found binlog data file '{}' left after an "
+                        "interrupted binlog purge",
+                        it->first);
+    garbage_object_names.push_back(it->first);
+    it = object_names.erase(it);
+  }
+}
+
+void storage_core::remove_garbage_objects(
+    const storage_object_name_list &garbage_object_names) {
+  if (garbage_object_names.empty()) {
+    return;
+  }
+
+  // in the querying_only mode the storage is never modified
+  if (construction_mode_ == storage_construction_mode_type::querying_only) {
+    return;
+  }
+
+  backend_->remove_objects(garbage_object_names);
+  logger_->log_format(
+      log_severity::warning,
+      "removed {} storage object(s) left after improper shutdown",
+      std::size(garbage_object_names));
+}
+
 void storage_core::initialize_storage_encryption(
     const optional_encryption_config &encryption_config) {
   if (encryption_config.has_value()) {
@@ -899,6 +1128,39 @@ void storage_core::save_binlog_metadata(const binlog_record &record) const {
                        util::as_const_byte_span(content));
 }
 
+void storage_core::recover_binlog_file_size(const binlog_record &record,
+                                            std::uint64_t actual_size,
+                                            bool is_last) {
+  // validating that the size stored in the metadata matches the actual size
+  if (record.size == actual_size) {
+    return;
+  }
+  // in case when Binlog Server process was not properly shut down
+  // there is a chance that there will be mismatch between the actual
+  // binlog data file size and the 'size' field in the binlog metadata
+
+  // if this mismatch was found in the most recent binlog file, we can
+  // perform automatic recovery (truncating binlog data file content to
+  // the size from the metadata)
+  if (!is_last) {
+    util::exception_location().raise<std::logic_error>(
+        "size from the binlog metadata does not match the actual binlog "
+        "size");
+  }
+  // performing recovery
+  if (record.size > actual_size) {
+    util::exception_location().raise<std::logic_error>(
+        "cannot perform recovery - size from the binlog metadata is "
+        "bigger than the actual binlog file size");
+  }
+  const auto binlog_file_name{record.name.str()};
+  backend_->resize_object(binlog_file_name, record.size);
+  logger_->log_format(log_severity::warning,
+                      "recovered binlog file '{}' by truncating it to "
+                      "the size from the metadata",
+                      binlog_file_name);
+}
+
 void storage_core::load_and_validate_binlog_metadata_set(
     // NOLINTNEXTLINE(bugprone-easily-swappable-parameters)
     const storage_object_name_container &object_names,
@@ -914,6 +1176,7 @@ void storage_core::load_and_validate_binlog_metadata_set(
             "missing metadata for a binlog listed in the binlog index");
       }
       loaded_binlog_metadata = load_binlog_metadata(record_it->name);
+      validate_binlog_metadata(loaded_binlog_metadata);
     } catch (const std::exception &) {
       if (construction_mode_ == storage_construction_mode_type::querying_only) {
         // in the querying_only mode we just skip invalid metadata and the
@@ -925,38 +1188,14 @@ void storage_core::load_and_validate_binlog_metadata_set(
       }
       throw;
     }
-    validate_binlog_metadata(loaded_binlog_metadata);
     // validating binlog size from the metadata only makes sense if we are not
     // in the querying_only mode
     if (construction_mode_ != storage_construction_mode_type::querying_only) {
-      const auto binlog_file_name{record_it->name.str()};
-      const auto actual_binlog_file_size{object_names.at(binlog_file_name)};
-      // validating that the size stored in the metadata matches the actual size
-      if (loaded_binlog_metadata.size != actual_binlog_file_size) {
-        // in case when Binlog Server process was not properly shut down
-        // there is a chance that there will be mismatch between the actual
-        // binlog data file size and the 'size' field in the binlog metadata
-
-        // if this mismatch was found in the most recent binlog file, we can
-        // perform automatic recovery (truncating binlog data file content to
-        // the size from the metadata)
-        if (std::next(record_it) != std::end(binlog_records_)) {
-          util::exception_location().raise<std::logic_error>(
-              "size from the binlog metadata does not match the actual binlog "
-              "size");
-        }
-        // performing recovery
-        if (loaded_binlog_metadata.size > actual_binlog_file_size) {
-          util::exception_location().raise<std::logic_error>(
-              "cannot perform recovery - size from the binlog metadata is "
-              "bigger than the actual binlog file size");
-        }
-        backend_->resize_object(binlog_file_name, loaded_binlog_metadata.size);
-        logger_->log_format(log_severity::warning,
-                            "recovered binlog file '{}' by truncating it to "
-                            "the size from the metadata",
-                            binlog_file_name);
-      }
+      const auto actual_binlog_file_size{
+          object_names.at(record_it->name.str())};
+      const bool is_last{std::next(record_it) == std::end(binlog_records_)};
+      recover_binlog_file_size(loaded_binlog_metadata, actual_binlog_file_size,
+                               is_last);
     }
     *record_it = std::move(loaded_binlog_metadata);
     ++record_it;
@@ -973,9 +1212,11 @@ void storage_core::load_and_validate_binlog_metadata_set(
 
   // if we are in GTID replication mode, then we can consider GTIDs from the
   // first binlog metadata as purged GTIDs for the whole storage
-  const auto &optional_added_gtids{binlog_records_.front().added_gtids};
-  if (optional_added_gtids.has_value()) {
-    purged_gtids_ = *optional_added_gtids;
+  if (!binlog_records_.empty()) {
+    const auto &optional_added_gtids{binlog_records_.front().added_gtids};
+    if (optional_added_gtids.has_value()) {
+      purged_gtids_ = *optional_added_gtids;
+    }
   }
 }
 
diff --git a/src/binsrv/storage_core.hpp b/src/binsrv/storage_core.hpp
index 2f2650a4..f36b94c4 100644
--- a/src/binsrv/storage_core.hpp
+++ b/src/binsrv/storage_core.hpp
@@ -272,7 +272,22 @@ class [[nodiscard]] storage_core {
     return binlog_records_.empty();
   }
 
+  using storage_object_name_list = std::vector<std::string>;
+
   void remove_temporary_objects(storage_object_name_container &object_names);
+  [[nodiscard]] bool collect_interrupted_first_binlog_objects(
+      storage_object_name_container &object_names,
+      storage_object_name_list &garbage_object_names) const;
+  void collect_interrupted_rotation_objects(
+      storage_object_name_container &object_names,
+      storage_object_name_container &object_metadata_names,
+      storage_object_name_list &garbage_object_names) const;
+  void collect_interrupted_purge_objects(
+      storage_object_name_container &object_names,
+      storage_object_name_container &object_metadata_names,
+      storage_object_name_list &garbage_object_names) const;
+  void
+  remove_garbage_objects(const storage_object_name_list &garbage_object_names);
 
   void initialize_storage_encryption(
       const optional_encryption_config &encryption_config);
@@ -341,6 +356,8 @@ class [[nodiscard]] storage_core {
   void validate_binlog_metadata(const binlog_record &record) const;
   void save_binlog_metadata(const binlog_record &record) const;
 
+  void recover_binlog_file_size(const binlog_record &record,
+                                std::uint64_t actual_size, bool is_last);
   void load_and_validate_binlog_metadata_set(
       const storage_object_name_container &object_names,
       const storage_object_name_container &object_metadata_names);
diff --git a/src/operations/purge_binlogs_operation.cpp b/src/operations/purge_binlogs_operation.cpp
index 5807306a..b3955d0e 100644
--- a/src/operations/purge_binlogs_operation.cpp
+++ b/src/operations/purge_binlogs_operation.cpp
@@ -65,8 +65,8 @@ generic_operation<mode_type::purge_binlogs>::execute() const {
     // call reports it via a non-empty 'cleanup_warning_message' and
     // we surface that as a 'warning' status (instead of plain
     // 'success'), with the underlying error message attached, so the
-    // operator knows the storage will need attention before the next
-    // 'fetch' / 'pull' run.
+    // operator knows that the leftovers will be removed as garbage on
+    // the next startup.
     binsrv::models::search_response response;
     if (!cleanup_warning_message.empty()) {
       response = binsrv::models::search_response{