diff --git a/CMakeLists.txt b/CMakeLists.txt
index 6529639..cee8444 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -33,6 +33,8 @@ include_directories(${SQLite3_INCLUDE_DIRS})
 add_executable(${PROJECT_NAME}
     src/main.cpp
     src/ServerInterface.cpp
+    src/FtpLogFetcher.cpp
+    src/LogEntryLister.cpp
     src/LogLoader.cpp)
 
 target_link_libraries(${PROJECT_NAME}
diff --git a/README.md b/README.md
index 8993abb..bdb04f4 100644
--- a/README.md
+++ b/README.md
@@ -4,9 +4,48 @@ Downloads PX4 log files (.ulg) and uploads them to a local server and optionally
 
 The **config.toml** file is used to configure the program settings.
 
+Works with PX4 (`.ulg`) and ArduPilot (`.BIN`).
+
 ### Behavior
 Downloading and uploading will only occur while the vehicle is not armed. Downloading and uploading operations are performed in separate threads. An sqlite database per server is used to track log file download/upload status.
 
+### Log transport
+The bulk transfer uses **MAVLink FTP** (`FILE_TRANSFER_PROTOCOL`, msg 110), not the classic log protocol (`LOG_DATA`, msg 120).
+
+`LOG_DATA` has no `target_system`/`target_component` fields, so a router in between — mavlink-router, for instance — has no addressing information to work with and copies every chunk to every endpoint it serves, telemetry radio included. `FILE_TRANSFER_PROTOCOL` is addressed, and both PX4 and ArduPilot send the reply back to the requesting sysid/compid, so the transfer is unicast to logloader and the other endpoints stay quiet. No router or autopilot configuration change is needed.
+
+The log *list* still comes from `LOG_ENTRY` (msg 118), which is also untargeted, but it is a handful of bytes per log rather than megabytes, and it is the only source of the timestamp the databases are keyed on.
+
+At startup logloader probes for the vehicle's log directory (`/fs/microsd/log` on PX4, `/APM/LOGS` on ArduPilot) and uses the result to pick the log file extension. Each downloaded file is size-checked against what `LOG_ENTRY` reported before it is moved into the logs directory; anything that does not line up is discarded and re-fetched over `LOG_DATA` instead.
+
+Requirements:
+- The autopilot's MAVLink instance must have FTP enabled (`mavlink start -x` on PX4).
+- If FTP is unavailable logloader falls back to `LOG_DATA` automatically. On ArduPilot without FTP, set `log_extension = ".BIN"` so downloaded files are named correctly.
+
+### ArduPilot notes
+ArduPilot needs FTP: MAVSDK's `LogFiles` plugin cannot enumerate its logs. The plugin assumes PX4's zero-based log ids — it discards any `LOG_ENTRY` whose id is not below `num_logs` and reads the collected entries back out at indices `0..num_logs-1` — while ArduPilot numbers list entries from one, so its final entry always trips that check and `get_entries()` returns `NoLogfiles`. When the FTP probe reports ArduPilot, logloader runs the `LOG_REQUEST_LIST` exchange itself (`LogEntryLister`) without assuming where the numbering starts.
+
+Mapping a list entry onto a file is also stack-specific. ArduPilot reports a *list entry number*, not the log number in the file name, and log numbers wrap at `LOG_MAX_FILES`. logloader reads `LASTLOG.TXT` over FTP and reproduces ArduPilot's own oldest-first ordering to resolve the two.
+
+`.BIN` files are not accepted by review.px4.io, so leave `upload_enabled = false` or point `remote_server` somewhere that understands dataflash logs.
+
+Set `download_protocol` in **config.toml** to `"mavlink"` to restore the old behavior, or `"ftp"` to never fall back.
+
+### Configuration
+| Key | Default | Description |
+| --- | --- | --- |
+| `connection_url` | `udp://:14551` | MAVSDK connection string |
+| `local_server` | `http://127.0.0.1:5006` | Local upload target |
+| `remote_server` | `https://review.px4.io` | Remote upload target |
+| `email` | `""` | Email attached to remote uploads |
+| `remote_api_key` | `""` | Per-account API key for authenticated Flight Review (`Authorization: Bearer` + `X-API-Key`). Empty = upload without API key headers (open servers). Generate under /account |
+| `upload_enabled` | `false` | Upload to the remote server |
+| `public_logs` | `false` | Mark remote uploads public |
+| `download_protocol` | `"auto"` | `"auto"`, `"ftp"` or `"mavlink"` |
+| `remote_log_directory` | `""` | Override the vehicle log directory |
+| `log_extension` | `""` | Override the downloaded file extension |
+| `ftp_use_burst` | `true` | Use FTP burst reads |
+
 ### Build
 Install dependencies
 ```
diff --git a/config.toml b/config.toml
index ff208c2..d1758b1 100644
--- a/config.toml
+++ b/config.toml
@@ -4,3 +4,31 @@ remote_server = "https://review.px4.io"
 email = ""
 upload_enabled = false
 public_logs = false
+
+# Per-account API key for authenticated Flight Review (ARK).
+# Generate at https://review.arkelectron.com/account (shown once; format fr_...).
+# When set: sent as Authorization: Bearer and X-API-Key on POST /upload.
+# When empty: remote uploads are still attempted without API key headers
+# (works for open servers; ARK Flight Review will reject with 403).
+remote_api_key = ""
+
+# How to pull the log data off the vehicle.
+#   "auto"    - MAVLink FTP when the vehicle answers, LOG_DATA otherwise (default)
+#   "ftp"     - MAVLink FTP only, never fall back to LOG_DATA
+#   "mavlink" - the classic LOG_DATA protocol only
+#
+# FTP is preferred because FILE_TRANSFER_PROTOCOL is addressed to this system and
+# component, so mavlink-router sends the transfer to this endpoint only. LOG_DATA has no
+# target fields, which leaves a router no choice but to copy every chunk to every
+# endpoint it serves, telemetry radio included.
+download_protocol = "auto"
+
+# Remote log directory. Empty probes "/fs/microsd/log" (PX4) then "/APM/LOGS" (ArduPilot).
+remote_log_directory = ""
+
+# Extension for downloaded logs. Empty derives it from the flight stack, which needs FTP.
+# Set it to ".BIN" for ArduPilot when FTP is unavailable.
+log_extension = ""
+
+# FTP burst reads. Much faster, disable it if the flight stack's burst support misbehaves.
+ftp_use_burst = true
diff --git a/docs/hardware-test-pr28.md b/docs/hardware-test-pr28.md
new file mode 100644
index 0000000..658fb2f
--- /dev/null
+++ b/docs/hardware-test-pr28.md
@@ -0,0 +1,71 @@
+## Hardware test results (Pi + PX4 / ARK Pi6X)
+
+Tested branch `claude/logdata-mavlink-targeting-uj8d0y` including `7a23c8e` (`fix(upload): stop spamming when flight-review is offline`) on a live companion:
+
+| Item | Value |
+|------|--------|
+| Host | Raspberry Pi (ark-os), aarch64 |
+| Autopilot | PX4 via `mavlink-router` UART @ 2 Mbaud |
+| Link | `udp://:14551` (logloader endpoint) |
+| Binary | Built from this branch with MAVSDK 3.17.1, `DEBUG_BUILD=ON` |
+
+### Connect / index
+
+```text
+Connected.
+MAVLink FTP available, PX4 logs in /fs/microsd/log (68 files)
+Received 68 log entries in ~0.26s
+```
+
+After FC reboot this was reliable. Earlier mid-session hard-kills of logloader left the FC with `FTP list … File Does Not Exist` / `0 log entries` until reboot.
+
+### FTP download (success)
+
+| Log | Protocol | Result |
+|-----|----------|--------|
+| `LOG0058` (~40 MB) | MAVLink FTP (burst) | **Finished in 94s** — on-disk file is valid ULog (`ULog\x01…`) |
+| Path | `/fs/microsd/log/2025-02-05/20_39_10.ulg` | Mapped and transferred |
+
+### Path mapping miss → auto fallback (success)
+
+For a small entry (`LOG0020`, LOG_ENTRY size **259471**):
+
+```text
+Downloading …/LOG0020_….ulg from /fs/microsd/log/2024-06-18/16_06_57.ulg
+Size mismatch …: got 276495 bytes, expected 259471
+FTP download failed
+Downloading …/LOG0020_….ulg          # LOG_DATA fallback (download_protocol=auto)
+Finished in 0.55 seconds
+```
+
+Final file size **259471**, valid ULog magic. So **auto** correctly falls back when FTP path resolution picks the wrong remote file.
+
+### Large-log FTP timeouts
+
+Newest/largest queue head (~100 MB, `LOG0067`) often hits:
+
+```text
+FTP download of …/00_02_04.ulg failed: Timeout
+```
+
+Burst mode makes progress (multi‑Mbps) but MAVSDK can still time out on very large files in short runs. Non-burst was ~800 kbps. Worth a follow-up (timeouts / resume / queue order), not a blocker for basic FTP path.
+
+### Upload spam fix (`7a23c8e`)
+
+With **flight-review not running** (`:5006` down) and multiple pending local uploads:
+
+| Metric | Count |
+|--------|------:|
+| `Upload server … unreachable; skipping uploads for 60s` | **1** |
+| `TEMPORARILY FAILED` (old spam) | **0** |
+| Per-log `Connection to … failed` | **0** |
+
+### Not tested here
+
+- ArduPilot / `.BIN` / `LogEntryLister` path  
+- Live upload to a running local flight-review (only “server down” path)  
+- Full multi-hour campaign of all 68 logs over FTP only  
+
+### Verdict
+
+**Works on PX4 hardware after a healthy FC boot:** connect, FTP detect/index, successful multi‑MB FTP download, auto→LOG_DATA fallback on size mismatch, and no upload spam when flight-review is offline.
diff --git a/src/FtpLogFetcher.cpp b/src/FtpLogFetcher.cpp
new file mode 100644
index 0000000..9f9baf5
--- /dev/null
+++ b/src/FtpLogFetcher.cpp
@@ -0,0 +1,565 @@
+#include "FtpLogFetcher.hpp"
+#include "Log.hpp"
+
+#include <algorithm>
+#include <cctype>
+#include <cstdlib>
+#include <ctime>
+#include <filesystem>
+#include <fstream>
+#include <future>
+#include <iomanip>
+#include <sstream>
+#include <type_traits>
+
+namespace fs = std::filesystem;
+
+namespace
+{
+
+struct Candidate {
+	const char* directory;
+	FtpLogFetcher::Flavor flavor;
+	const char* extension;
+};
+
+// PX4 keeps ulog files under PX4_STORAGEDIR "/log", ArduPilot keeps dataflash logs under
+// HAL_BOARD_STORAGE_DIRECTORY "/LOGS". Both are reachable over MAVLink FTP.
+constexpr Candidate kCandidates[] = {
+	{"/fs/microsd/log", FtpLogFetcher::Flavor::PX4, ".ulg"},
+	{"/APM/LOGS", FtpLogFetcher::Flavor::ArduPilot, ".BIN"},
+};
+
+// PX4 nests logs one level deep (log/<date>/<time>.ulg), ArduPilot keeps them flat.
+constexpr int kMaxListDepth = 2;
+
+struct Listing {
+	std::vector<std::string> directories;
+	std::vector<std::pair<std::string, std::optional<uint32_t>>> files;
+};
+
+// MAVSDK 2.x returns the raw protocol entries, MAVSDK 3.x splits them up and drops the
+// file sizes. Both shapes have to compile against whichever version is installed.
+template <typename T>
+Listing flatten_listing(const T& listing)
+{
+	Listing out;
+
+	if constexpr(std::is_same_v<T, std::vector<std::string>>) {
+		for (const auto& raw : listing) {
+			if (raw.size() < 2) {
+				continue;
+			}
+
+			const std::string body = raw.substr(1);
+
+			if (raw[0] == 'D') {
+				out.directories.push_back(body);
+
+			} else if (raw[0] == 'F') {
+				const auto tab = body.find('\t');
+
+				if (tab == std::string::npos) {
+					out.files.emplace_back(body, std::nullopt);
+					continue;
+				}
+
+				std::optional<uint32_t> size_bytes;
+
+				try {
+					size_bytes = static_cast<uint32_t>(std::stoul(body.substr(tab + 1)));
+
+				} catch (const std::exception&) {
+					size_bytes = std::nullopt;
+				}
+
+				out.files.emplace_back(body.substr(0, tab), size_bytes);
+			}
+		}
+
+	} else {
+		for (const auto& directory : listing.dirs) {
+			out.directories.push_back(directory);
+		}
+
+		for (const auto& file : listing.files) {
+			out.files.emplace_back(file, std::nullopt);
+		}
+	}
+
+	return out;
+}
+
+bool ends_with_ignore_case(const std::string& text, const std::string& suffix)
+{
+	if (text.size() < suffix.size()) {
+		return false;
+	}
+
+	const size_t offset = text.size() - suffix.size();
+
+	for (size_t i = 0; i < suffix.size(); i++) {
+		if (std::tolower(text[offset + i]) != std::tolower(suffix[i])) {
+			return false;
+		}
+	}
+
+	return true;
+}
+
+std::optional<int64_t> parse_iso8601_utc(const std::string& text)
+{
+	std::tm tm {};
+	std::istringstream ss(text);
+	ss >> std::get_time(&tm, "%Y-%m-%dT%H:%M:%SZ");
+
+	if (ss.fail()) {
+		return std::nullopt;
+	}
+
+	return static_cast<int64_t>(timegm(&tm));
+}
+
+// PX4 writes logs as <root>/<YYYY-MM-DD>/<HH_MM_SS>.ulg
+std::optional<int64_t> parse_px4_start_time(const std::string& path)
+{
+	const fs::path parsed(path);
+	const std::string day = parsed.parent_path().filename().string();
+	const std::string time_of_day = parsed.stem().string();
+
+	if (day.size() != 10 || time_of_day.size() != 8) {
+		return std::nullopt;
+	}
+
+	std::string iso = day + "T" + time_of_day + "Z";
+	std::replace(iso.begin(), iso.end(), '_', ':');
+
+	return parse_iso8601_utc(iso);
+}
+
+// ArduPilot writes logs as <root>/<%08u>.BIN
+std::optional<uint32_t> parse_ardupilot_log_number(const std::string& path)
+{
+	const std::string stem = fs::path(path).stem().string();
+
+	if (stem.empty() || stem.size() > 9) {
+		return std::nullopt;
+	}
+
+	for (char c : stem) {
+		if (std::isdigit(static_cast<unsigned char>(c)) == 0) {
+			return std::nullopt;
+		}
+	}
+
+	return static_cast<uint32_t>(std::stoul(stem));
+}
+
+// ArduPilot lists logs oldest first while the file names carry the log number, and log
+// numbers wrap around. Once they have, every number above the most recent one belongs to
+// the older half of the list. Returns positions into log_numbers, oldest first.
+std::vector<size_t> ardupilot_list_order(const std::vector<std::optional<uint32_t>>& log_numbers,
+		std::optional<uint32_t> last_log_number)
+{
+	std::vector<size_t> numbered;
+
+	for (size_t i = 0; i < log_numbers.size(); i++) {
+		if (log_numbers[i].has_value()) {
+			numbered.push_back(i);
+		}
+	}
+
+	if (numbered.empty()) {
+		return {};
+	}
+
+	std::sort(numbered.begin(), numbered.end(), [&log_numbers](size_t lhs, size_t rhs) {
+		return log_numbers[lhs].value() < log_numbers[rhs].value();
+	});
+
+	const uint32_t last = last_log_number.value_or(log_numbers[numbered.back()].value());
+
+	std::vector<size_t> order;
+
+	for (size_t i : numbered) {
+		if (log_numbers[i].value() > last) {
+			order.push_back(i);
+		}
+	}
+
+	for (size_t i : numbered) {
+		if (log_numbers[i].value() <= last) {
+			order.push_back(i);
+		}
+	}
+
+	return order;
+}
+
+std::string join_path(const std::string& directory, const std::string& name)
+{
+	if (!directory.empty() && directory.back() == '/') {
+		return directory + name;
+	}
+
+	return directory + "/" + name;
+}
+
+} // namespace
+
+FtpLogFetcher::FtpLogFetcher(std::shared_ptr<mavsdk::System> system, const FtpLogFetcher::Settings& settings)
+	: _ftp(std::make_shared<mavsdk::Ftp>(system))
+	, _settings(settings)
+{
+	std::error_code ec;
+	fs::create_directories(_settings.temp_directory, ec);
+}
+
+void FtpLogFetcher::stop()
+{
+	_should_exit = true;
+}
+
+std::string FtpLogFetcher::flavor_name() const
+{
+	switch (_flavor) {
+	case Flavor::PX4: return "PX4";
+
+	case Flavor::ArduPilot: return "ArduPilot";
+
+	default: return "unknown";
+	}
+}
+
+std::string FtpLogFetcher::log_extension() const
+{
+	return _extension;
+}
+
+bool FtpLogFetcher::detect()
+{
+	// A flight stack that answers with an empty listing rather than an error would let the
+	// wrong candidate win, so take a directory that actually holds logs over one that
+	// merely exists. A vehicle with no logs yet falls through to the second pass.
+	for (bool require_logs : {true, false}) {
+		for (const auto& candidate : kCandidates) {
+			const std::string directory = _settings.remote_directory.empty()
+						      ? std::string(candidate.directory)
+						      : _settings.remote_directory;
+
+			if (!build_index(directory, candidate.extension, candidate.flavor)) {
+				continue;
+			}
+
+			if (require_logs && _index.empty()) {
+				continue;
+			}
+
+			_flavor = candidate.flavor;
+			_remote_directory = directory;
+			_extension = candidate.extension;
+
+			LOG("MAVLink FTP available, " << flavor_name() << " logs in " << _remote_directory
+			    << " (" << _index.size() << " files)");
+
+			return true;
+		}
+	}
+
+	LOG("MAVLink FTP unavailable, falling back to the MAVLink log protocol");
+
+	return false;
+}
+
+bool FtpLogFetcher::refresh()
+{
+	if (_flavor == Flavor::Unknown) {
+		// The SD card may have shown up after we connected
+		return detect();
+	}
+
+	return build_index(_remote_directory, _extension, _flavor);
+}
+
+bool FtpLogFetcher::build_index(const std::string& root, const std::string& extension, Flavor flavor)
+{
+	_index.clear();
+	_ardupilot_order.clear();
+
+	if (!list_recursive(root, kMaxListDepth, extension)) {
+		return false;
+	}
+
+	if (flavor == Flavor::ArduPilot) {
+		build_ardupilot_order(root);
+	}
+
+	return true;
+}
+
+bool FtpLogFetcher::list_recursive(const std::string& directory, int depth_remaining, const std::string& extension)
+{
+	if (depth_remaining <= 0 || _should_exit) {
+		return true;
+	}
+
+	auto response = _ftp->list_directory(directory);
+
+	if (response.first != mavsdk::Ftp::Result::Success) {
+		LOG_DEBUG("FTP list of " << directory << " failed: " << response.first);
+		return false;
+	}
+
+	const Listing listing = flatten_listing(response.second);
+
+	for (const auto& [name, size_bytes] : listing.files) {
+		if (!ends_with_ignore_case(name, extension)) {
+			continue;
+		}
+
+		RemoteFile file;
+		file.path = join_path(directory, name);
+		file.size_bytes = size_bytes;
+		file.start_time = parse_px4_start_time(file.path);
+		file.log_number = parse_ardupilot_log_number(file.path);
+		_index.push_back(file);
+	}
+
+	for (const auto& name : listing.directories) {
+		if (name == "." || name == "..") {
+			continue;
+		}
+
+		// A subdirectory we cannot read is not fatal, the logs we did find are still usable
+		list_recursive(join_path(directory, name), depth_remaining - 1, extension);
+	}
+
+	return true;
+}
+
+void FtpLogFetcher::build_ardupilot_order(const std::string& root)
+{
+	std::vector<std::optional<uint32_t>> log_numbers;
+	log_numbers.reserve(_index.size());
+
+	bool any_numbered = false;
+
+	for (const auto& file : _index) {
+		log_numbers.push_back(file.log_number);
+		any_numbered = any_numbered || file.log_number.has_value();
+	}
+
+	if (!any_numbered) {
+		return;
+	}
+
+	_ardupilot_order = ardupilot_list_order(log_numbers, read_last_log_number(root));
+}
+
+std::optional<uint32_t> FtpLogFetcher::read_last_log_number(const std::string& root)
+{
+	const std::string remote_path = join_path(root, "LASTLOG.TXT");
+	// Deliberately not LASTLOG.TXT, that is where the transfer is staged
+	const std::string local_path = join_path(_settings.temp_directory, "last_log_number.txt");
+
+	std::error_code ec;
+	fs::remove(local_path, ec);
+
+	// Burst mode is pointless for a handful of bytes and not every build supports it
+	if (!download(remote_path, local_path, 0, nullptr)) {
+		LOG_DEBUG("No LASTLOG.TXT, assuming the log numbers have not wrapped");
+		return std::nullopt;
+	}
+
+	std::ifstream file(local_path);
+	uint32_t last_log_number = 0;
+	const bool parsed = static_cast<bool>(file >> last_log_number);
+	file.close();
+	fs::remove(local_path, ec);
+
+	if (!parsed || last_log_number == 0) {
+		return std::nullopt;
+	}
+
+	return last_log_number;
+}
+
+std::string FtpLogFetcher::resolve(const mavsdk::LogFiles::Entry& entry)
+{
+	// An exact size match is unambiguous when MAVSDK reports sizes, which is the common
+	// case. Otherwise fall back to reproducing the ordering the flight stack listed.
+	RemoteFile* file = match_by_size(entry.size_bytes);
+
+	if (file == nullptr) {
+		file = (_flavor == Flavor::ArduPilot) ? match_ardupilot(entry) : match_px4(entry);
+	}
+
+	if (file == nullptr) {
+		return "";
+	}
+
+	file->claimed = true;
+
+	return file->path;
+}
+
+FtpLogFetcher::RemoteFile* FtpLogFetcher::match_by_size(uint32_t size_bytes)
+{
+	if (size_bytes == 0) {
+		return nullptr;
+	}
+
+	RemoteFile* match = nullptr;
+
+	for (auto& file : _index) {
+		if (file.claimed || file.size_bytes.value_or(0) != size_bytes) {
+			continue;
+		}
+
+		if (match != nullptr) {
+			// Two logs of exactly the same size, let the caller's fallback sort it out
+			return nullptr;
+		}
+
+		match = &file;
+	}
+
+	return match;
+}
+
+FtpLogFetcher::RemoteFile* FtpLogFetcher::match_px4(const mavsdk::LogFiles::Entry& entry)
+{
+	// PX4 reports the file mtime in LOG_ENTRY, which is when the log was last written to.
+	// The log that was open at that moment is the newest one started at or before it.
+	const auto modified_time = parse_iso8601_utc(entry.date);
+
+	if (!modified_time.has_value()) {
+		return nullptr;
+	}
+
+	RemoteFile* match = nullptr;
+
+	for (auto& file : _index) {
+		if (file.claimed || !file.start_time.has_value() || file.start_time.value() > modified_time.value()) {
+			continue;
+		}
+
+		if (match == nullptr || file.start_time.value() > match->start_time.value()) {
+			match = &file;
+		}
+	}
+
+	return match;
+}
+
+FtpLogFetcher::RemoteFile* FtpLogFetcher::match_ardupilot(const mavsdk::LogFiles::Entry& entry)
+{
+	// ArduPilot reports a list entry number, not the log number in the file name
+	if (entry.id == 0 || entry.id > _ardupilot_order.size()) {
+		return nullptr;
+	}
+
+	RemoteFile* file = &_index[_ardupilot_order[entry.id - 1]];
+
+	return file->claimed ? nullptr : file;
+}
+
+void FtpLogFetcher::release(const std::string& remote_path)
+{
+	for (auto& file : _index) {
+		if (file.path == remote_path) {
+			file.claimed = false;
+			return;
+		}
+	}
+}
+
+bool FtpLogFetcher::download(const std::string& remote_path, const std::string& local_path, uint32_t expected_size,
+			     const FtpLogFetcher::ProgressCallback& progress)
+{
+	std::error_code ec;
+	fs::create_directories(_settings.temp_directory, ec);
+
+	// MAVSDK writes to <local_dir>/<remote basename>, so stage the transfer and move it
+	// into place afterwards. A partial file never appears in the logs directory.
+	const std::string staged_path = join_path(_settings.temp_directory, fs::path(remote_path).filename().string());
+	fs::remove(staged_path, ec);
+
+	struct Transfer {
+		std::promise<mavsdk::Ftp::Result> promise;
+		std::atomic<bool> settled {false};
+	};
+
+	// Shared so the callback stays valid if we bail out early on shutdown
+	auto transfer = std::make_shared<Transfer>();
+	auto future_result = transfer->promise.get_future();
+
+	_ftp->download_async(remote_path, _settings.temp_directory, _settings.use_burst,
+	[transfer, progress, this](mavsdk::Ftp::Result result, mavsdk::Ftp::ProgressData data) {
+		if (transfer->settled) {
+			return;
+		}
+
+		if (_should_exit) {
+			transfer->settled = true;
+			transfer->promise.set_value(mavsdk::Ftp::Result::Timeout);
+			return;
+		}
+
+		if (result == mavsdk::Ftp::Result::Next) {
+			if (progress) {
+				progress(data.bytes_transferred, data.total_bytes);
+			}
+
+			return;
+		}
+
+		transfer->settled = true;
+		transfer->promise.set_value(result);
+	});
+
+	const auto result = future_result.get();
+
+	if (result != mavsdk::Ftp::Result::Success) {
+		LOG_DEBUG("FTP download of " << remote_path << " failed: " << result);
+		fs::remove(staged_path, ec);
+		return false;
+	}
+
+	if (!fs::exists(staged_path)) {
+		LOG("FTP reported success but " << staged_path << " is missing");
+		return false;
+	}
+
+	// The vehicle picked the file, we picked which entry it belongs to. The size is the
+	// only thing tying the two together, so refuse anything that does not line up.
+	const uint32_t downloaded_size = static_cast<uint32_t>(fs::file_size(staged_path, ec));
+
+	if (expected_size != 0 && downloaded_size != expected_size) {
+		LOG("Size mismatch for " << remote_path << ": got " << downloaded_size << " bytes, expected " << expected_size);
+		fs::remove(staged_path, ec);
+		return false;
+	}
+
+	if (staged_path == local_path) {
+		return true;
+	}
+
+	fs::remove(local_path, ec);
+	fs::rename(staged_path, local_path, ec);
+
+	if (ec) {
+		// Different filesystems, fall back to a copy
+		ec.clear();
+		fs::copy_file(staged_path, local_path, fs::copy_options::overwrite_existing, ec);
+		std::error_code remove_ec;
+		fs::remove(staged_path, remove_ec);
+
+		if (ec) {
+			LOG("Failed to move " << staged_path << " to " << local_path << ": " << ec.message());
+			return false;
+		}
+	}
+
+	return true;
+}
diff --git a/src/FtpLogFetcher.hpp b/src/FtpLogFetcher.hpp
new file mode 100644
index 0000000..b27495f
--- /dev/null
+++ b/src/FtpLogFetcher.hpp
@@ -0,0 +1,101 @@
+#pragma once
+
+#include <mavsdk/system.h>
+#include <mavsdk/plugins/ftp/ftp.h>
+#include <mavsdk/plugins/log_files/log_files.h>
+
+#include <atomic>
+#include <cstdint>
+#include <functional>
+#include <memory>
+#include <optional>
+#include <string>
+#include <vector>
+
+// Downloads flight logs over MAVLink FTP instead of the classic log protocol.
+//
+// LOG_DATA (msg 120) carries no target_system/target_component, so a router such as
+// mavlink-router has to treat it as a broadcast and copies every chunk to every endpoint
+// it serves -- including the telemetry radio. FILE_TRANSFER_PROTOCOL (msg 110) is
+// addressed, and both PX4 and ArduPilot send the reply back to the requesting
+// sysid/compid, so the bulk transfer is unicast to us and the other endpoints stay quiet.
+//
+// The log list still comes from LOG_ENTRY (msg 118). That is also untargeted, but it is
+// a handful of bytes per log rather than megabytes, and it is the only source of the
+// timestamp we key the databases on.
+class FtpLogFetcher
+{
+public:
+	enum class Flavor {
+		Unknown,
+		PX4,
+		ArduPilot
+	};
+
+	struct Settings {
+		std::string temp_directory;   // Scratch space for in-progress transfers
+		std::string remote_directory; // Empty means probe for it
+		bool use_burst {true};
+	};
+
+	using ProgressCallback = std::function<void(uint32_t bytes_transferred, uint32_t total_bytes)>;
+
+	FtpLogFetcher(std::shared_ptr<mavsdk::System> system, const Settings& settings);
+
+	// Probes the vehicle for a log directory and builds the initial index.
+	// Call once after connecting, before any download is attempted.
+	bool detect();
+
+	// Rebuilds the remote file index and forgets previous path claims.
+	bool refresh();
+
+	bool available() const { return _flavor != Flavor::Unknown; }
+	Flavor flavor() const { return _flavor; }
+	const std::string& remote_directory() const { return _remote_directory; }
+	std::string flavor_name() const;
+	std::string log_extension() const;
+
+	// Maps a log entry onto a remote path. Returns an empty string when the entry cannot
+	// be matched, in which case the caller should fall back to the LOG_DATA protocol.
+	std::string resolve(const mavsdk::LogFiles::Entry& entry);
+
+	// Downloads remote_path and moves it onto local_path once the size checks out.
+	bool download(const std::string& remote_path, const std::string& local_path, uint32_t expected_size,
+		      const ProgressCallback& progress);
+
+	// Hands a path back to the index after a failed download so a retry can pick it up.
+	void release(const std::string& remote_path);
+
+	void stop();
+
+private:
+	struct RemoteFile {
+		std::string path;
+		std::optional<uint32_t> size_bytes; // MAVSDK >= 3 does not report sizes
+		std::optional<int64_t> start_time;  // PX4: parsed out of the path
+		std::optional<uint32_t> log_number; // ArduPilot: parsed out of the file name
+		bool claimed {false};
+	};
+
+	bool build_index(const std::string& root, const std::string& extension, Flavor flavor);
+	bool list_recursive(const std::string& directory, int depth_remaining, const std::string& extension);
+	void build_ardupilot_order(const std::string& root);
+
+	RemoteFile* match_by_size(uint32_t size_bytes);
+	RemoteFile* match_px4(const mavsdk::LogFiles::Entry& entry);
+	RemoteFile* match_ardupilot(const mavsdk::LogFiles::Entry& entry);
+
+	std::optional<uint32_t> read_last_log_number(const std::string& root);
+
+	std::shared_ptr<mavsdk::Ftp> _ftp;
+	Settings _settings;
+
+	Flavor _flavor {Flavor::Unknown};
+	std::string _remote_directory;
+	std::string _extension;
+
+	std::vector<RemoteFile> _index;
+	std::vector<size_t> _ardupilot_order; // Index positions, oldest log first
+
+	std::atomic<bool> _should_exit {false};
+};
diff --git a/src/LogEntryLister.cpp b/src/LogEntryLister.cpp
new file mode 100644
index 0000000..3cd1a5e
--- /dev/null
+++ b/src/LogEntryLister.cpp
@@ -0,0 +1,181 @@
+#include "LogEntryLister.hpp"
+#include "Log.hpp"
+
+#include <chrono>
+#include <ctime>
+
+namespace
+{
+
+// How long to wait for another LOG_ENTRY before assuming the reply stalled
+constexpr auto kQuietPeriod = std::chrono::seconds(3);
+
+// How many times to re-ask for the entries we are still missing
+constexpr int kMaxAttempts = 4;
+
+std::string iso8601_utc(uint32_t time_utc)
+{
+	// Matches the formatting MAVSDK's LogFiles plugin uses, so the uuids the databases
+	// are keyed on stay the same no matter which listing path produced the entry.
+	char buffer[sizeof "2018-08-31T20:50:42Z"];
+	const time_t as_time_t = time_utc;
+	strftime(buffer, sizeof(buffer), "%FT%TZ", gmtime(&as_time_t));
+	return buffer;
+}
+
+} // namespace
+
+LogEntryLister::LogEntryLister(std::shared_ptr<mavsdk::System> system)
+	: _passthrough(std::make_shared<mavsdk::MavlinkPassthrough>(system))
+{
+	_subscription = _passthrough->subscribe_message(MAVLINK_MSG_ID_LOG_ENTRY,
+	[this](const mavlink_message_t& message) {
+		handle_log_entry(message);
+	});
+}
+
+LogEntryLister::~LogEntryLister()
+{
+	_passthrough->unsubscribe_message(MAVLINK_MSG_ID_LOG_ENTRY, _subscription);
+}
+
+void LogEntryLister::stop()
+{
+	_should_exit = true;
+	_cv.notify_all();
+}
+
+void LogEntryLister::handle_log_entry(const mavlink_message_t& message)
+{
+	mavlink_log_entry_t log_entry;
+	mavlink_msg_log_entry_decode(&message, &log_entry);
+
+	{
+		std::lock_guard<std::mutex> lock(_mutex);
+
+		_num_logs = log_entry.num_logs;
+		_last_log_id = log_entry.last_log_num;
+		_got_any = true;
+
+		if (log_entry.num_logs > 0) {
+			mavsdk::LogFiles::Entry entry;
+			entry.id = log_entry.id;
+			entry.date = iso8601_utc(log_entry.time_utc);
+			entry.size_bytes = log_entry.size;
+			_entries[log_entry.id] = entry;
+		}
+	}
+
+	_cv.notify_all();
+}
+
+bool LogEntryLister::listing_complete() const
+{
+	if (!_got_any) {
+		return false;
+	}
+
+	if (_num_logs == 0) {
+		return true;
+	}
+
+	// PX4 numbers from zero and ArduPilot from one, so derive the range the vehicle is
+	// actually offering rather than assuming either.
+	if (_last_log_id + 1 < _num_logs) {
+		return false;
+	}
+
+	const uint16_t first_id = _last_log_id + 1 - _num_logs;
+
+	for (uint16_t id = first_id; id <= _last_log_id; id++) {
+		if (_entries.find(id) == _entries.end()) {
+			return false;
+		}
+	}
+
+	return true;
+}
+
+void LogEntryLister::request_list(uint16_t start, uint16_t end)
+{
+	_passthrough->queue_message([this, start, end](MavlinkAddress address, uint8_t channel) {
+		mavlink_message_t message;
+		mavlink_msg_log_request_list_pack_chan(address.system_id, address.component_id, channel, &message,
+						       _passthrough->get_target_sysid(), _passthrough->get_target_compid(),
+						       start, end);
+		return message;
+	});
+}
+
+void LogEntryLister::request_end()
+{
+	_passthrough->queue_message([this](MavlinkAddress address, uint8_t channel) {
+		mavlink_message_t message;
+		mavlink_msg_log_request_end_pack_chan(address.system_id, address.component_id, channel, &message,
+						      _passthrough->get_target_sysid(), _passthrough->get_target_compid());
+		return message;
+	});
+}
+
+bool LogEntryLister::get_entries(std::vector<mavsdk::LogFiles::Entry>& entries)
+{
+	entries.clear();
+
+	{
+		std::lock_guard<std::mutex> lock(_mutex);
+		_entries.clear();
+		_num_logs = 0;
+		_last_log_id = 0;
+		_got_any = false;
+	}
+
+	uint16_t start = 0;
+
+	for (int attempt = 0; attempt < kMaxAttempts && !_should_exit; attempt++) {
+		request_list(start, 0xFFFF);
+
+		// Wait for the reply to finish or go quiet. Every entry that lands resets the
+		// clock, so a slow link gets as long as it needs.
+		std::unique_lock<std::mutex> lock(_mutex);
+
+		while (!_should_exit && !listing_complete()) {
+			if (_cv.wait_for(lock, kQuietPeriod) == std::cv_status::timeout) {
+				break;
+			}
+		}
+
+		if (_should_exit) {
+			return false;
+		}
+
+		if (listing_complete()) {
+			for (const auto& [id, entry] : _entries) {
+				entries.push_back(entry);
+			}
+
+			lock.unlock();
+			request_end();
+			return true;
+		}
+
+		// Pick up from the first gap rather than replaying the whole list
+		if (_got_any && _num_logs > 0 && _last_log_id + 1 >= _num_logs) {
+			const uint16_t first_id = _last_log_id + 1 - _num_logs;
+			start = first_id;
+
+			for (uint16_t id = first_id; id <= _last_log_id; id++) {
+				if (_entries.find(id) == _entries.end()) {
+					start = id;
+					break;
+				}
+			}
+		}
+
+		LOG_DEBUG("Log listing incomplete, have " << _entries.size() << " of " << _num_logs
+			  << ", retrying from " << start);
+	}
+
+	request_end();
+
+	return false;
+}
diff --git a/src/LogEntryLister.hpp b/src/LogEntryLister.hpp
new file mode 100644
index 0000000..baa816b
--- /dev/null
+++ b/src/LogEntryLister.hpp
@@ -0,0 +1,56 @@
+#pragma once
+
+#include <mavsdk/system.h>
+#include <mavsdk/plugins/log_files/log_files.h>
+#include <mavsdk/plugins/mavlink_passthrough/mavlink_passthrough.h>
+
+#include <atomic>
+#include <condition_variable>
+#include <cstdint>
+#include <map>
+#include <memory>
+#include <mutex>
+#include <vector>
+
+// Enumerates the logs on the vehicle with LOG_REQUEST_LIST / LOG_ENTRY.
+//
+// MAVSDK's LogFiles plugin bakes in PX4's zero-based numbering: it discards any LOG_ENTRY
+// whose id is not below num_logs, and it reads the collected entries back out at indices
+// 0..num_logs-1. ArduPilot numbers list entries from one, so its final entry always trips
+// that check and get_entries() reports NoLogfiles. This runs the same exchange without
+// assuming where the numbering starts.
+//
+// Only the list travels this way. The bulk transfer goes over MAVLink FTP, see
+// FtpLogFetcher.
+class LogEntryLister
+{
+public:
+	explicit LogEntryLister(std::shared_ptr<mavsdk::System> system);
+	~LogEntryLister();
+
+	// Blocks until the vehicle has listed every log or we give up on it.
+	bool get_entries(std::vector<mavsdk::LogFiles::Entry>& entries);
+
+	void stop();
+
+private:
+	void handle_log_entry(const mavlink_message_t& message);
+	void request_list(uint16_t start, uint16_t end);
+	void request_end();
+
+	// True once every id the vehicle told us to expect has arrived. Caller holds _mutex.
+	bool listing_complete() const;
+
+	std::shared_ptr<mavsdk::MavlinkPassthrough> _passthrough;
+	mavsdk::MavlinkPassthrough::MessageHandle _subscription;
+
+	mutable std::mutex _mutex;
+	std::condition_variable _cv;
+
+	std::map<uint16_t, mavsdk::LogFiles::Entry> _entries;
+	uint16_t _num_logs {0};
+	uint16_t _last_log_id {0};
+	bool _got_any {false};
+
+	std::atomic<bool> _should_exit {false};
+};
diff --git a/src/LogLoader.cpp b/src/LogLoader.cpp
index 357b0fe..b2f4d30 100644
--- a/src/LogLoader.cpp
+++ b/src/LogLoader.cpp
@@ -27,6 +27,7 @@ LogLoader::LogLoader(const LogLoader::Settings& settings)
 		.db_path = _settings.application_directory + "local_server.db",
 		.upload_enabled = true, // Always upload to local server
 		.public_logs = true, // Public required true for searching using Web UI
+		.api_key = "", // Local Flight Review is typically open on the companion
 	};
 
 	// Setup remote server interface
@@ -37,6 +38,7 @@ LogLoader::LogLoader(const LogLoader::Settings& settings)
 		.db_path = _settings.application_directory + "remote_server.db",
 		.upload_enabled = settings.upload_enabled,
 		.public_logs = settings.public_logs,
+		.api_key = settings.remote_api_key,
 	};
 
 	_local_server = std::make_shared<ServerInterface>(local_server_settings);
@@ -54,6 +56,14 @@ void LogLoader::stop()
 		_should_exit = true;
 	}
 	_exit_cv.notify_all();
+
+	if (_ftp_fetcher) {
+		_ftp_fetcher->stop();
+	}
+
+	if (_log_entry_lister) {
+		_log_entry_lister->stop();
+	}
 }
 
 bool LogLoader::wait_for_mavsdk_connection(double timeout_ms)
@@ -80,10 +90,60 @@ bool LogLoader::wait_for_mavsdk_connection(double timeout_ms)
 	// MAVSDK plugins
 	_log_files = std::make_shared<mavsdk::LogFiles>(system.value());
 	_telemetry = std::make_shared<mavsdk::Telemetry>(system.value());
+	_log_entry_lister = std::make_shared<LogEntryLister>(system.value());
+
+	setup_ftp_fetcher(system.value());
 
 	return true;
 }
 
+void LogLoader::setup_ftp_fetcher(std::shared_ptr<mavsdk::System> system)
+{
+	_ftp_fetcher.reset();
+
+	if (_settings.download_protocol == "mavlink") {
+		LOG("Using the MAVLink log protocol (LOG_DATA) as configured");
+
+	} else {
+		FtpLogFetcher::Settings ftp_settings = {
+			.temp_directory = _settings.application_directory + "tmp/",
+			.remote_directory = _settings.remote_log_directory,
+			.use_burst = _settings.ftp_use_burst,
+		};
+
+		auto fetcher = std::make_shared<FtpLogFetcher>(system, ftp_settings);
+
+		if (fetcher->detect()) {
+			_ftp_fetcher = fetcher;
+
+		} else if (_settings.download_protocol == "ftp") {
+			// Pinned to FTP, keep the fetcher around and let it retry as we run
+			LOG("download_protocol is \"ftp\" but the vehicle did not answer, will keep trying");
+			_ftp_fetcher = fetcher;
+		}
+	}
+
+	apply_log_extension();
+}
+
+void LogLoader::apply_log_extension()
+{
+	// The log protocol gives us no way to tell PX4 and ArduPilot apart, so an operator
+	// running ArduPilot without FTP has to name the extension in the config file.
+	std::string extension = _settings.log_extension;
+
+	if (extension.empty() && _ftp_fetcher && _ftp_fetcher->available()) {
+		extension = _ftp_fetcher->log_extension();
+	}
+
+	if (extension.empty()) {
+		extension = ".ulg";
+	}
+
+	_local_server->set_log_extension(extension);
+	_remote_server->set_log_extension(extension);
+}
+
 void LogLoader::run()
 {
 	auto upload_thread = std::thread(&LogLoader::upload_logs_thread, this);
@@ -125,7 +185,12 @@ void LogLoader::run()
 		while (!_should_exit && num_remaining) {
 			// Download logs until we should exit or there are none left to download
 			LOG("Downloading log " << total_to_download - num_remaining + 1 << "/" << total_to_download);
-			download_next_log();
+
+			if (!download_next_log()) {
+				// Nothing was retired, so retrying right away would just spin
+				break;
+			}
+
 			num_remaining = _local_server->num_logs_to_download();
 		}
 
@@ -146,17 +211,25 @@ bool LogLoader::request_log_entries()
 
 	// Debug profiling code. We need to check how this performs with 100+ logs
 	auto request_start = std::chrono::high_resolution_clock::now();
-	auto entries_result = _log_files->get_entries();
 
-	//  Store log entries
-	_log_entries = entries_result.second;
+	bool success = false;
+
+	if (_ftp_fetcher && _ftp_fetcher->flavor() == FtpLogFetcher::Flavor::ArduPilot) {
+		// MAVSDK's LogFiles plugin cannot enumerate ArduPilot logs, see LogEntryLister
+		success = _log_entry_lister->get_entries(_log_entries);
+
+	} else {
+		auto entries_result = _log_files->get_entries();
+		_log_entries = entries_result.second;
+		success = entries_result.first == mavsdk::LogFiles::Result::Success;
+	}
 
 	auto request_end = std::chrono::high_resolution_clock::now();
 
 	std::chrono::duration<double> request_duration = request_end - request_start;
 	LOG_DEBUG("Received " << _log_entries.size() << "log entries in " << request_duration.count() << " seconds");
 
-	if (entries_result.first != mavsdk::LogFiles::Result::Success) {
+	if (!success) {
 		LOG("Error getting log entries");
 		return false;
 	}
@@ -175,16 +248,29 @@ bool LogLoader::request_log_entries()
 	LOG_DEBUG("Added log entries to databases in " << db_duration.count() << " seconds");
 	LOG_DEBUG("Total processing time: " << (request_duration + db_duration).count() << " seconds");
 
+	// Listing the remote directory costs a round trip per subdirectory, so only do it
+	// when there is actually something to fetch.
+	if (_ftp_fetcher && _local_server->num_logs_to_download()) {
+		if (_ftp_fetcher->refresh()) {
+			apply_log_extension();
+
+		} else {
+			LOG_DEBUG("Could not index the vehicle log directory over FTP");
+		}
+	}
+
 	return true;
 }
 
-void LogLoader::download_next_log()
+// Returns true when an entry was retired from the download queue, false when the caller
+// should back off instead of immediately trying again.
+bool LogLoader::download_next_log()
 {
 	// Get one undownloaded log, use the local server for query
 	ServerInterface::DatabaseEntry db_entry = _local_server->get_next_log_to_download();
 
 	if (db_entry.uuid.empty()) {
-		return;
+		return false;
 	}
 
 	// Find the corresponding log entry in the list from the vehicle
@@ -193,13 +279,15 @@ void LogLoader::download_next_log()
 		std::string uuid = ServerInterface::generate_uuid(entry);
 
 		if (uuid == db_entry.uuid) {
-			if (download_log(entry)) {
-				// Update downloaded status in both databases
-				_local_server->update_download_status(uuid, true);
-				_remote_server->update_download_status(uuid, true);
+			if (!download_log(entry)) {
+				return false;
 			}
 
-			return;
+			// Update downloaded status in both databases
+			_local_server->update_download_status(uuid, true);
+			_remote_server->update_download_status(uuid, true);
+
+			return true;
 		}
 	}
 
@@ -209,13 +297,11 @@ void LogLoader::download_next_log()
 	_local_server->update_download_status(db_entry.uuid, true);
 	_remote_server->update_download_status(db_entry.uuid, true);
 
-	return;
+	return true;
 }
 
 bool LogLoader::download_log(const mavsdk::LogFiles::Entry& entry)
 {
-	auto prom = std::promise<mavsdk::LogFiles::Result> {};
-	auto future_result = prom.get_future();
 	auto download_path = _local_server->filepath_from_entry(entry);
 
 	// Check and delete file if it already exists. This can occur due to partial download.
@@ -231,6 +317,79 @@ bool LogLoader::download_log(const mavsdk::LogFiles::Entry& entry)
 		}
 	}
 
+	// Prefer MAVLink FTP. Unlike LOG_DATA it is addressed to us, so a router in between
+	// unicasts the transfer instead of copying every chunk to every endpoint it serves.
+	if (_ftp_fetcher && _ftp_fetcher->available()) {
+		if (download_log_ftp(entry, download_path)) {
+			return true;
+		}
+	}
+
+	if (_settings.download_protocol == "ftp") {
+		LOG("Could not fetch " << download_path << " over FTP and download_protocol is \"ftp\", not falling back");
+		return false;
+	}
+
+	return download_log_mavlink(entry, download_path);
+}
+
+bool LogLoader::download_log_ftp(const mavsdk::LogFiles::Entry& entry, const std::string& download_path)
+{
+	const std::string remote_path = _ftp_fetcher->resolve(entry);
+
+	if (remote_path.empty()) {
+		LOG_DEBUG("No remote path matches log " << entry.id << " (" << entry.date << ")");
+		return false;
+	}
+
+	LOG("Downloading " << download_path << " from " << remote_path);
+
+	auto time_start = std::chrono::steady_clock::now();
+
+	// Everything the callback touches is captured by value, MAVSDK may still hold it
+	// after we have stopped waiting on the transfer.
+	bool success = _ftp_fetcher->download(remote_path, download_path, entry.size_bytes,
+	[date = entry.date, size_bytes = entry.size_bytes, time_start](uint32_t bytes_transferred, uint32_t total_bytes) {
+#ifdef DEBUG_BUILD
+		auto elapsed_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
+					  std::chrono::steady_clock::now() - time_start).count();
+		double rate_kbps = elapsed_ms > 0 ? (bytes_transferred * 8.0) / elapsed_ms : 0.;
+		int percent = total_bytes > 0 ? int((100.0 * bytes_transferred) / total_bytes) : 0;
+
+		LOG_DEBUG("Downloading: "
+			  << std::setw(24) << std::left << date
+			  << std::setw(8) << std::fixed << std::setprecision(2) << size_bytes / 1e6 << "MB"
+			  << std::setw(6) << std::right << percent << "%"
+			  << std::setw(12) << std::fixed << std::setprecision(2) << rate_kbps << " Kbps"
+			  << std::flush);
+#else
+		(void)date;
+		(void)size_bytes;
+		(void)time_start;
+		(void)bytes_transferred;
+		(void)total_bytes;
+#endif
+	});
+
+	if (!success) {
+		// Let a later pass try this file again rather than stranding it
+		_ftp_fetcher->release(remote_path);
+		LOG("FTP download failed");
+		return false;
+	}
+
+	double seconds = std::chrono::duration_cast<std::chrono::milliseconds>(
+				 std::chrono::steady_clock::now() - time_start).count() / 1000.;
+	LOG("Finished in " << std::setprecision(2) << seconds << " seconds");
+
+	return true;
+}
+
+bool LogLoader::download_log_mavlink(const mavsdk::LogFiles::Entry& entry, const std::string& download_path)
+{
+	auto prom = std::promise<mavsdk::LogFiles::Result> {};
+	auto future_result = prom.get_future();
+
 	LOG("Downloading " << download_path);
 
 	auto time_start = std::chrono::steady_clock::now();
@@ -344,12 +503,25 @@ void LogLoader::upload_pending_logs(std::shared_ptr<ServerInterface> server)
 		if (result.success) {
 			LOG("Log upload SUCCESS: " << result.message);
 
-		} else if (result.status_code == 400) {
-			LOG("Log upload failed (" << result.status_code << "): " << result.message);
+		} else if (result.status_code == 400 || result.status_code == 401 || result.status_code == 403) {
+			LOG("Log upload rejected (" << result.status_code << "): " << result.message);
+
+			// Account/auth policy applies to every log on this server — stop the batch.
+			if (result.status_code == 401 || result.status_code == 403) {
+				LOG("Stopping remote uploads for this cycle (server requires login/approval)");
+				return;
+			}
+
+		} else if (result.status_code == 503) {
+			// Server down (e.g. local flight-review not running). Already logged once
+			// with a cooldown in ServerInterface; do not walk the rest of the queue.
+			return;
 
 		} else {
 			LOG("Log upload TEMPORARILY FAILED (" << result.status_code << "): "
 			    << result.message << " - Will retry later");
+			// Don't tight-loop on transient errors for the same file.
+			return;
 		}
 	}
 }
diff --git a/src/LogLoader.hpp b/src/LogLoader.hpp
index 0f2df34..fdb5c47 100644
--- a/src/LogLoader.hpp
+++ b/src/LogLoader.hpp
@@ -6,6 +6,8 @@
 #include <mavsdk/log_callback.h>
 #include <condition_variable>
 
+#include "FtpLogFetcher.hpp"
+#include "LogEntryLister.hpp"
 #include "ServerInterface.hpp"
 
 class LogLoader
@@ -15,10 +17,18 @@ class LogLoader
 		std::string email;
 		std::string local_server;
 		std::string remote_server;
+		// API key for remote_server (ARK Flight Review account key). Empty = none.
+		std::string remote_api_key;
 		std::string mavsdk_connection_url;
 		std::string application_directory;
 		bool upload_enabled;
 		bool public_logs;
+		// "auto" prefers MAVLink FTP and falls back to the log protocol, "ftp" and
+		// "mavlink" pin one of the two.
+		std::string download_protocol;
+		std::string remote_log_directory; // Empty means probe for it
+		std::string log_extension;        // Empty means derive it from the flight stack
+		bool ftp_use_burst;
 	};
 
 	LogLoader(const Settings& settings);
@@ -28,10 +38,15 @@ class LogLoader
 	bool wait_for_mavsdk_connection(double timeout_ms);
 
 private:
+	void setup_ftp_fetcher(std::shared_ptr<mavsdk::System> system);
+	void apply_log_extension();
+
 	// Download
 	bool request_log_entries();
-	void download_next_log();
+	bool download_next_log();
 	bool download_log(const mavsdk::LogFiles::Entry& entry);
+	bool download_log_ftp(const mavsdk::LogFiles::Entry& entry, const std::string& download_path);
+	bool download_log_mavlink(const mavsdk::LogFiles::Entry& entry, const std::string& download_path);
 
 	// Upload
 	void upload_logs_thread();
@@ -47,6 +62,8 @@ class LogLoader
 	std::shared_ptr<mavsdk::Mavsdk> _mavsdk;
 	std::shared_ptr<mavsdk::Telemetry> _telemetry;
 	std::shared_ptr<mavsdk::LogFiles> _log_files;
+	std::shared_ptr<FtpLogFetcher> _ftp_fetcher;
+	std::shared_ptr<LogEntryLister> _log_entry_lister;
 	std::vector<mavsdk::LogFiles::Entry> _log_entries;
 
 	std::atomic<bool> _should_exit = false;
diff --git a/src/ServerInterface.cpp b/src/ServerInterface.cpp
index 554893b..d10f16c 100644
--- a/src/ServerInterface.cpp
+++ b/src/ServerInterface.cpp
@@ -278,9 +278,9 @@ ServerInterface::UploadResult ServerInterface::upload_log(const std::string& fil
 			sqlite3_finalize(stmt);
 		}
 
-	} else if (result.status_code == 400) {
-		// Permanent failure - add to blacklist
-		add_to_blacklist(uuid, "HTTP 400: Bad Request");
+	} else if (result.status_code == 400 || result.status_code == 401 || result.status_code == 403) {
+		// Permanent failure - do not keep retrying the same file
+		add_to_blacklist(uuid, "HTTP " + std::to_string(result.status_code) + ": " + result.message);
 	}
 
 	return result;
@@ -356,10 +356,23 @@ ServerInterface::DatabaseEntry ServerInterface::get_next_log_to_download()
 	return entry;
 }
 
+void ServerInterface::set_log_extension(const std::string& extension)
+{
+	std::lock_guard<std::mutex> lock(_log_extension_mutex);
+	_log_extension = extension;
+}
+
+std::string ServerInterface::log_extension() const
+{
+	std::lock_guard<std::mutex> lock(_log_extension_mutex);
+	return _log_extension;
+}
+
 std::string ServerInterface::filepath_from_entry(const mavsdk::LogFiles::Entry& entry) const
 {
 	std::ostringstream ss;
-	ss << _settings.logs_directory << "LOG" << std::setfill('0') << std::setw(4) << entry.id << "_" << entry.date << ".ulg";
+	ss << _settings.logs_directory << "LOG" << std::setfill('0') << std::setw(4) << entry.id << "_" << entry.date <<
+	   log_extension();
 	return ss.str();
 }
 
@@ -386,8 +399,20 @@ std::string ServerInterface::filepath_from_uuid(const std::string& uuid) const
 		if (date_text != nullptr) {
 			std::string date = reinterpret_cast<const char*>(date_text);
 			std::ostringstream ss;
-			ss << _settings.logs_directory << "LOG" << std::setfill('0') << std::setw(4) << id << "_" << date << ".ulg";
-			filepath = ss.str();
+			ss << _settings.logs_directory << "LOG" << std::setfill('0') << std::setw(4) << id << "_" << date;
+			const std::string base = ss.str();
+
+			filepath = base + log_extension();
+
+			if (!fs::exists(filepath)) {
+				// Logs downloaded before the flight stack was known may carry another extension
+				for (const char* extension : {".ulg", ".BIN", ".bin"}) {
+					if (fs::exists(base + extension)) {
+						filepath = base + extension;
+						break;
+					}
+				}
+			}
 		}
 	}
 
@@ -413,7 +438,9 @@ ServerInterface::UploadResult ServerInterface::upload(const std::string& filepat
 	}
 
 	if (!server_reachable()) {
-		return {false, 0, "Server unreachable: " + _settings.server_url};
+		// 503: treat as service unavailable so the upload loop can bail the batch
+		// instead of hammering a dead local flight-review for every pending log.
+		return {false, 503, "Server unreachable: " + _settings.server_url};
 	}
 
 	std::ifstream file(filepath, std::ios::binary);
@@ -438,25 +465,63 @@ ServerInterface::UploadResult ServerInterface::upload(const std::string& filepat
 	std::string content((std::istreambuf_iterator<char>(file)), std::istreambuf_iterator<char>());
 	items.push_back({"filearg", content, filepath, "application/octet-stream"});
 
-	LOG("Uploading " << fs::path(filepath).filename().string() << " to " << _settings.server_url);
+	// API-key headers only when a non-empty key is configured. Never send
+	// Authorization/X-API-Key with an empty value.
+	httplib::Headers headers;
+	const bool use_api_key = !_settings.api_key.empty();
+
+	if (use_api_key) {
+		// ARK Flight Review (flight_review api_key.py):
+		//   Authorization: Bearer <key>
+		//   X-API-Key: <key>
+		// (Query ?api_key= also works server-side; form fields cannot — auth
+		// runs in prepare() before the body is available.)
+		headers.emplace("Authorization", "Bearer " + _settings.api_key);
+		headers.emplace("X-API-Key", _settings.api_key);
+	}
+
+	LOG("Uploading " << fs::path(filepath).filename().string() << " to " << _settings.server_url
+	    << (use_api_key ? " (with API key)" : ""));
 
 	// Post multi-part form
 	httplib::Result res;
 
 	if (_protocol == Protocol::Https) {
 		httplib::SSLClient cli(_settings.server_url);
-		res = cli.Post("/upload", items);
+		cli.set_connection_timeout(30, 0);
+		cli.set_read_timeout(120, 0);
+		res = use_api_key ? cli.Post("/upload", headers, items)
+		      : cli.Post("/upload", items);
 
 	} else {
 		httplib::Client cli(_settings.server_url);
-		res = cli.Post("/upload", items);
+		cli.set_connection_timeout(30, 0);
+		cli.set_read_timeout(120, 0);
+		res = use_api_key ? cli.Post("/upload", headers, items)
+		      : cli.Post("/upload", items);
 	}
 
 	if (res && res->status == 302) {
 		return {true, 302, "Success: " + _settings.server_url + res->get_header_value("Location")};
 
-	} else if (res && res->status == 400) {
-		return {false, 400, "Bad Request - Will not retry"};
+	} else if (res && (res->status == 400 || res->status == 401 || res->status == 403)) {
+		// Permanent client/auth errors: do not spin on the same log forever.
+		// review.arkelectron.com returns 403 when the account is not logged in /
+		// not approved for automated uploads.
+		std::string detail = res->body;
+		// Collapse HTML / whitespace for a one-line log message.
+		for (char& c : detail) {
+			if (c == '\n' || c == '\r' || c == '\t') {
+				c = ' ';
+			}
+		}
+
+		if (detail.size() > 200) {
+			detail.resize(200);
+			detail += "...";
+		}
+
+		return {false, res->status, detail.empty() ? "Rejected by server (will not retry)" : detail};
 
 	} else {
 		return {false, res ? res->status : 0, "Will retry later"};
@@ -465,24 +530,58 @@ ServerInterface::UploadResult ServerInterface::upload(const std::string& filepat
 
 bool ServerInterface::server_reachable()
 {
+	const auto now = std::chrono::steady_clock::now();
+
+	// Still inside the cooldown from a previous failure — do not re-probe or re-log.
+	if (now < _unreachable_until) {
+		return false;
+	}
+
 	httplib::Result res;
 
 	if (_protocol == Protocol::Https) {
 		httplib::SSLClient cli(_settings.server_url);
+		cli.set_connection_timeout(5, 0);
+		cli.set_read_timeout(5, 0);
+		// Flight Review often redirects "/" (302). We only need proof the host answers.
+		cli.set_follow_location(false);
 		res = cli.Get("/");
 
 	} else {
 		httplib::Client cli(_settings.server_url);
+		cli.set_connection_timeout(5, 0);
+		cli.set_read_timeout(5, 0);
+		cli.set_follow_location(false);
 		res = cli.Get("/");
 	}
 
-	bool success = res && res->status == 200;
+	// Any HTTP response means the server is up. Flight Review's home page returns
+	// 302, so requiring status == 200 falsely marked review.arkelectron.com dead.
+	// Connection / TLS failures leave res empty.
+	const bool success = static_cast<bool>(res);
 
 	if (!success) {
-		LOG("Connection to " << _settings.server_url << " failed: " << (res ? std::to_string(res->status) : "No response"));
+		_unreachable_until = now + kUnreachableCooldown;
+
+		if (!_reported_unreachable) {
+			const std::string detail = res ? ("HTTP " + std::to_string(res->status)) : "No response";
+			LOG("Upload server " << _settings.server_url << " unreachable (" << detail
+			    << "); skipping uploads for "
+			    << std::chrono::duration_cast<std::chrono::seconds>(kUnreachableCooldown).count()
+			    << "s");
+			_reported_unreachable = true;
+		}
+
+		return false;
 	}
 
-	return success;
+	if (_reported_unreachable) {
+		LOG("Upload server " << _settings.server_url << " is reachable again");
+		_reported_unreachable = false;
+	}
+
+	_unreachable_until = {};
+	return true;
 }
 
 bool ServerInterface::init_database()
diff --git a/src/ServerInterface.hpp b/src/ServerInterface.hpp
index ea47559..804cda9 100644
--- a/src/ServerInterface.hpp
+++ b/src/ServerInterface.hpp
@@ -1,5 +1,7 @@
 #pragma once
 
+#include <chrono>
+#include <mutex>
 #include <string>
 #include <vector>
 #include <sqlite3.h>
@@ -15,11 +17,19 @@ class ServerInterface
 		std::string db_path;         // Path to this server's database
 		bool upload_enabled {};
 		bool public_logs {};
+		// Per-account API key for authenticated Flight Review instances
+		// (e.g. review.arkelectron.com). Sent as:
+		//   Authorization: Bearer <key>
+		//   X-API-Key: <key>
+		// Matches ARK flight_review api_key.py. Leave empty for open servers.
+		std::string api_key;
 	};
 
 	struct UploadResult {
 		bool success;
-		int status_code;    // HTTP status code, or 0 if not applicable
+		// HTTP status, or 0 when not applicable. 503 means the server is unreachable
+		// (connectivity); callers should stop the current upload batch and retry later.
+		int status_code;
 		std::string message;
 	};
 
@@ -56,6 +66,10 @@ class ServerInterface
 	std::string filepath_from_entry(const mavsdk::LogFiles::Entry& entry) const ;
 	std::string filepath_from_uuid(const std::string& uuid) const;
 
+	// ".ulg" for PX4, ".BIN" for ArduPilot. Set once the flight stack is known.
+	void set_log_extension(const std::string& extension);
+	std::string log_extension() const;
+
 	void start();
 	void stop();
 
@@ -78,4 +92,15 @@ class ServerInterface
 	Protocol _protocol {Protocol::Https};
 	bool _should_exit = false;
 	sqlite3* _db = nullptr;
+
+	// Read by the upload thread while the download loop may still be setting it
+	mutable std::mutex _log_extension_mutex;
+	std::string _log_extension {".ulg"};
+
+	// When flight-review (or any upload target) is down, avoid probing and logging
+	// once per pending log. Probe at most every kUnreachableCooldown, and log only on
+	// the down/up transitions.
+	static constexpr auto kUnreachableCooldown = std::chrono::seconds(60);
+	std::chrono::steady_clock::time_point _unreachable_until {};
+	bool _reported_unreachable {false};
 };
diff --git a/src/main.cpp b/src/main.cpp
index 0621092..6e951c2 100644
--- a/src/main.cpp
+++ b/src/main.cpp
@@ -57,12 +57,38 @@ int main(int argc, char** argv)
 		.email = config["email"].value_or(""),
 		.local_server = config["local_server"].value_or("http://127.0.0.1:5006"),
 		.remote_server = config["remote_server"].value_or("https://logs.px4.io"),
+		.remote_api_key = config["remote_api_key"].value_or(""),
 		.mavsdk_connection_url = config["connection_url"].value_or("0.0.0"),
 		.application_directory = config["application_directory"].value_or(data_dir.string() + "/"),
 		.upload_enabled = config["upload_enabled"].value_or(false),
-		.public_logs = config["public_logs"].value_or(false)
+		.public_logs = config["public_logs"].value_or(false),
+		.download_protocol = config["download_protocol"].value_or("auto"),
+		.remote_log_directory = config["remote_log_directory"].value_or(""),
+		.log_extension = config["log_extension"].value_or(""),
+		.ftp_use_burst = config["ftp_use_burst"].value_or(true)
 	};
 
+	// Trim whitespace-only keys so they count as "not set" (no empty auth headers).
+	auto trim = [](std::string s) {
+		const auto start = s.find_first_not_of(" \t\r\n");
+
+		if (start == std::string::npos) {
+			return std::string{};
+		}
+
+		const auto end = s.find_last_not_of(" \t\r\n");
+		return s.substr(start, end - start + 1);
+	};
+	settings.remote_api_key = trim(std::move(settings.remote_api_key));
+
+	// Still attempt remote uploads without a key (open servers like logs.px4.io).
+	// Authenticated ARK Flight Review will 403 until remote_api_key is set; we
+	// never send empty Authorization / X-API-Key headers (see ServerInterface).
+	if (settings.upload_enabled && settings.remote_api_key.empty()) {
+		LOG("upload_enabled is true but remote_api_key is empty — remote uploads will "
+		    "proceed without an API key (open servers only; ARK Flight Review needs a key)");
+	}
+
 	_log_loader = std::make_shared<LogLoader>(settings);
 
 	bool connected = false;