Own the map download progress lifecycle

[FrogAi]

The upstream settings handler starts a download worker before recording ownership, so two commands arriving before the next progress poll can start overlapping downloads. The handler now claims ownership before launching the worker to close that gap.

Each run gets its own cancellation channel, published progress is copied, and the terminal update is retained. Idle or late cancellation cannot carry into a later run, and the UI can observe completion even when an older progress update is queued.

This follows the existing serial settings owner, single worker, and capacity-one progress queue. Cancellation still takes effect between archives; an in-flight archive finishes first.

Root cause, ownership, reproducible checks

Root cause

On upstream 7201c6b, the main loop handles openpilot and CLI input serially before polling progress. Handle(download) checks downloadActive and launches a worker, but that flag is only updated when GetDownloadProgress() consumes a message. Both commands can therefore see an idle downloader and launch workers sharing temporary files, output paths, progress, and cancellation.

The cancellation channel also outlives individual runs, so an idle or late token can remain for the next run. Sending DownloadProgress copies slice and map headers, leaving nested location details shared with the worker. Later changes can mutate a retained snapshot, while a full one-slot queue can hide the final Active=false message behind an older active update.

Change and rationale

The existing settings owner sets downloadActive=true and creates a buffered cancellation channel before starting the worker. It ignores another download command until it consumes terminal progress and releases the channel. Keeping ownership through that handoff prevents a new run from sharing the old progress slot.

flowchart LR
  I[Idle] -->|Download command| C[Claim ownership and create cancel channel]
  C --> W[Worker downloads archives and publishes snapshots]
  W -->|Completion or cancellation at archive boundary| T[Queue inactive terminal snapshot]
  T -->|Main loop consumes terminal progress| I

Loading

publishProgress() copies the location list, details map, and each detail value. It removes any stale queued snapshot and makes a nonblocking send. Under the actual single-producer, capacity-one contract, the newest snapshot, including completion, remains available. Intermediate updates are still coalesced; progress is not a lossless event history. No manager, mutex, or extra goroutine is needed.

Verification

Checked head 8e5e677, tree e435907c7ce22a6e75ce571d757609300a94abf5, against upstream 7201c6b4b4ec1b0b9ea21daa8c05b80fdd7e01ee. The production diff is confined to settings/settings.go and settings/download.go.

The complete five-test reproducer below invokes real Handle and Download. Its replacement HTTP transport blocks at controlled request boundaries and returns valid empty tar/gzip responses. Paths are temporary, and the fixture fails visibly if device output/custom-menu paths exist. It neither starts the daemon nor sends an external HTTP request.

Case	Upstream base	Current implementation
First HTTP request reached before any progress poll	Ownership still false	Ownership already true; repeat command ignored
Cancel while idle	Token remains queued	No token queued
Cancel before the first archive with an occupied progress slot	Stale active update hides cancellation	Inactive canceled terminal update, 0 downloads and no HTTP request
Serialize a retained snapshot after the next location starts	Earlier count/map changes	Snapshot remains byte-identical
Complete two archives with a delayed reader	Stale active update can hide completion	Inactive terminal update reports 2 downloads
Cancel during the first of two archives	No second request, but stale active progress hides cancellation	First archive finishes; terminal reports canceled with 1 download; no second request
Cancel after the final archive starts, then start another run	Shared channel lifetime permits a late token to escape	Final archive succeeds; consuming terminal releases ownership; the new run uses a fresh empty channel

The upstream late-token row follows the shared channel lifetime in source; the current test exercises that token and the subsequent run. The last two rows describe the cancellation boundary, not in-flight interruption. A cancellation after the final archive starts does not retroactively mark that successful completion canceled. Direct callers of Download remain responsible for coordinating their own calls and channels.

Executed on 2026-09-05 with Linux amd64 and Go 1.25.1. Fresh upstream runs reproduced missing synchronous ownership, idle-cancel leakage, snapshot mutation, and lost terminal progress. All five tests passed on the current implementation. The current focused race check also passed, including serialization of a retained snapshot while the worker advances; package vet passed.

The actual run used cached modules with GOPROXY=off, GOSUMDB=off, and container networking disabled. These are synthetic ownership/publication checks, not a device download session or an extraction audit. HTTP behavior, extraction, URLs, commands, and IPC fields are unchanged; this PR adds no in-flight HTTP cancellation or request timeout.

Integration with #136

This PR and #136 overlap in the downloader loop. The combined resolution must retain #136's selected-row iteration and call this PR's publishProgress() inside that loop, before the shared cancellation/download body. Choosing either whole side would drop the other change.

Combined validation: exact source tree a4c306906627db3ac7a8ab768651c8628d55465a combines #101 9d61f06a1288ec4ea6f74f7d56a3057444316a13, #103 20e7c25b054b6399360676a7f539a39b4fbf855c, #105 bfcfe77be066634e36054327b20cfa6541063b54, #107 8e5e677d1196838069e9665d4e9d962bcc1e116b, #116 6fd5bbd6cf617c24a7fefd5e302fd36688a1a63b, #136 30e8ce98ea7a4c8401dbb5bfc62120c84fc689e4. The only overlapping file is settings/download.go; the resolution retains #136's selected-row loop and #107's progress publication inside it.

On 2026-09-05, combined Linux amd64 tests (including the scratch regression fixtures), race checks, vet and build passed. Under ARM64 emulation, the existing Makefile build stage (make GO_CAPNP_PATH=/usr/local/go-capnp/std), committed repository tests, vet and both CLI help commands passed with Go 1.25.1; go.mod/go.sum stayed unchanged and the resulting executable is AArch64. The ARM64 run does not include the extra amd64 scratch tests. It used an isolated retained build image, not a new dependency-install/image rebuild or physical device. No production archive payload or live params were accessed. #105 still requires runtime-first rollout before regenerated tiles are distributed.

Reproduce from a fresh clone

Save the two complete blocks below beside each other as reproducer_test.go and reproduce.sh. Requirements are Linux, Bash, Git, tar, and Go 1.25.1. Use an isolated environment without /data/media/0 or /data/openpilot/mapd_download_menu.json. The runner archives each pinned revision into a separate temporary directory, adds only the test fixture, and leaves raw results in its printed output directory.

Populate a module cache from the pinned revisions before the runner disables dependency fetching. The setup command downloads repository/module dependencies; the tests use only the synthetic transport:

git clone https://github.com/FrogAi/mapd.git mapd
export GOTOOLCHAIN=local
export GOMODCACHE="$(go env GOMODCACHE)"
for commit in 7201c6b4b4ec1b0b9ea21daa8c05b80fdd7e01ee 8e5e677d1196838069e9665d4e9d962bcc1e116b; do
  seed=$(mktemp -d)
  git -C mapd archive "$commit" | tar -x -C "$seed"
  (cd "$seed" && go mod download)
done
bash reproduce.sh mapd

reproduce.sh:

#!/usr/bin/env bash
set -euo pipefail
repository=$(cd "$1" && pwd)
fixture_dir=$(cd "$(dirname "$0")" && pwd)
output_dir=${PUBLIC_OUTPUT:-$(mktemp -d "${TMPDIR:-/tmp}/mapd-pr107.XXXXXX")}
mkdir -p "$output_dir"
output_dir=$(cd "$output_dir" && pwd)
export GOTOOLCHAIN=local GOPROXY=off GOSUMDB=off
export GOCACHE="$output_dir/build-cache" GOPATH="$output_dir/gopath"
mkdir -p "$GOCACHE" "$GOPATH"
test "$(go env GOVERSION)" = go1.25.1
go version

for entry in upstream:7201c6b4b4ec1b0b9ea21daa8c05b80fdd7e01ee current:8e5e677d1196838069e9665d4e9d962bcc1e116b; do
  name=${entry%%:*}
  commit=${entry#*:}
  source="$output_dir/$name"
  mkdir "$source"
  git -C "$repository" archive "$commit" | tar -x -C "$source"
  cp "$fixture_dir/reproducer_test.go" "$source/settings/public_reproducer_test.go"

  cd "$source"
  set +e
  go test -mod=readonly -buildvcs=false -count=1 -timeout=90s -v -run 'TestPublicDownload' ./settings > "$output_dir/$name.txt" 2>&1
  status=$?
  set -e
  printf '%s %s exit=%s\n' "$name" "$commit" "$status"
  if [ "$name" = upstream ]; then test "$status" -eq 1; else test "$status" -eq 0; fi
done
cd "$output_dir/current"
go test -mod=readonly -buildvcs=false -race -count=1 -timeout=90s -run 'TestPublicDownload' ./settings > "$output_dir/current-race.txt" 2>&1
go vet -mod=readonly ./settings > "$output_dir/current-vet.txt" 2>&1
printf 'current race/vet passed\nresults: %s\n' "$output_dir"

reproducer_test.go:

package settings

import (
	"archive/tar"
	"bytes"
	"compress/gzip"
	"encoding/json"
	"io"
	"net/http"
	"os"
	"testing"
	"time"

	"capnproto.org/go/capnp/v3"
	"pfeifer.dev/mapd/cereal/custom"
)

type downloadTestTransport func(*http.Request) (*http.Response, error)

func (transport downloadTestTransport) RoundTrip(request *http.Request) (*http.Response, error) {
	return transport(request)
}

func setupDownloadTest(t *testing.T, requests chan<- struct{}, releases <-chan struct{}) {
	t.Helper()
	for _, path := range []string{"/data/media/0", "/data/openpilot/mapd_download_menu.json"} {
		if _, err := os.Stat(path); !os.IsNotExist(err) {
			t.Fatalf("test requires an isolated environment without %s: %v", path, err)
		}
	}
	t.Chdir(t.TempDir())
	previousMenu := boundingBoxesJson
	boundingBoxesJson = []byte(`{"test":{
		"first":{"full_name":"First","bounding_box":{"min_lat":0,"min_lon":0,"max_lat":2,"max_lon":2}},
		"second":{"full_name":"Second","bounding_box":{"min_lat":2,"min_lon":0,"max_lat":4,"max_lon":2}}
	}}`)
	var archive bytes.Buffer
	compressed := gzip.NewWriter(&archive)
	writer := tar.NewWriter(compressed)
	if err := writer.Close(); err != nil {
		t.Fatal(err)
	}
	if err := compressed.Close(); err != nil {
		t.Fatal(err)
	}
	previousClient := http.DefaultClient
	http.DefaultClient = &http.Client{Transport: downloadTestTransport(func(request *http.Request) (*http.Response, error) {
		requests <- struct{}{}
		<-releases
		return &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(bytes.NewReader(archive.Bytes())), Header: make(http.Header)}, nil
	})}
	t.Cleanup(func() {
		boundingBoxesJson = previousMenu
		http.DefaultClient = previousClient
	})
}

func TestPublicDownloadRetainsTerminalProgress(t *testing.T) {
	requests := make(chan struct{}, 1)
	releases := make(chan struct{})
	close(releases)
	setupDownloadTest(t, requests, releases)
	progress := make(chan DownloadProgress, 1)
	cancel := make(chan bool, 1)
	cancel <- true
	Download("test.first", progress, cancel)
	terminal := <-progress
	if terminal.Active || !terminal.Canceled || terminal.DownloadedFiles != 0 || terminal.TotalFiles != 1 {
		t.Fatalf("terminal progress was lost behind an active update: %+v", terminal)
	}
	if len(requests) != 0 {
		t.Fatal("pre-file cancellation made an HTTP request")
	}
}

func TestPublicDownloadSnapshotsRemainStable(t *testing.T) {
	requests := make(chan struct{}, 2)
	releases := make(chan struct{})
	setupDownloadTest(t, requests, releases)
	progress := make(chan DownloadProgress, 1)
	finished := make(chan struct{})
	go func() {
		Download("test.first,test.second", progress, nil)
		close(finished)
	}()
	defer func() {
		close(releases)
		<-finished
	}()
	waitDownloadRequest(t, requests)
	snapshot := <-progress
	before, err := json.Marshal(snapshot)
	if err != nil {
		t.Fatal(err)
	}
	releases <- struct{}{}
	waitDownloadRequest(t, requests)
	after, err := json.Marshal(snapshot)
	if err != nil || !bytes.Equal(before, after) {
		t.Errorf("published snapshot changed: before %s, after %s, error %v", before, after, err)
	}
	stopReading := make(chan struct{})
	readerFinished := make(chan struct{})
	go func() {
		defer close(readerFinished)
		for {
			select {
			case <-stopReading:
				return
			default:
				json.Marshal(snapshot)
			}
		}
	}()
	releases <- struct{}{}
	<-finished
	close(stopReading)
	<-readerFinished
	terminal := <-progress
	if terminal.Active || terminal.Canceled || terminal.DownloadedFiles != 2 || terminal.LocationDetails["test.second"].DownloadedFiles != 1 {
		t.Errorf("wrong completion progress: %+v", terminal)
	}
	after, err = json.Marshal(snapshot)
	if err != nil || !bytes.Equal(before, after) {
		t.Errorf("completion mutated retained snapshot: before %s, after %s, error %v", before, after, err)
	}
}

func TestPublicDownloadCancellationFinishesCurrentFile(t *testing.T) {
	requests := make(chan struct{}, 2)
	releases := make(chan struct{})
	setupDownloadTest(t, requests, releases)
	progress := make(chan DownloadProgress, 1)
	cancel := make(chan bool, 1)
	finished := make(chan struct{})
	go func() {
		Download("test.first,test.second", progress, cancel)
		close(finished)
	}()
	defer func() {
		close(releases)
		<-finished
	}()
	waitDownloadRequest(t, requests)
	cancel <- true
	releases <- struct{}{}
	<-finished
	terminal := <-progress
	if terminal.Active || !terminal.Canceled || terminal.DownloadedFiles != 1 || len(requests) != 0 {
		t.Fatalf("cancellation crossed the current-file boundary: %+v, extra requests %d", terminal, len(requests))
	}
}

func TestPublicDownloadIgnoresIdleCancellation(t *testing.T) {
	settings := MapdSettings{downloadProgress: make(chan DownloadProgress, 1), cancelDownload: make(chan bool, 1)}
	settings.Handle(downloadInput(t, custom.MapdInputType_cancelDownload, ""))
	if len(settings.cancelDownload) != 0 {
		t.Fatal("idle cancel left a token for a later download")
	}
}

func TestPublicDownloadSettingsOwnRun(t *testing.T) {
	requests := make(chan struct{}, 4)
	releases := make(chan struct{})
	setupDownloadTest(t, requests, releases)
	settings := MapdSettings{downloadProgress: make(chan DownloadProgress, 1), cancelDownload: make(chan bool, 1)}
	input := downloadInput(t, custom.MapdInputType_download, "test.first")
	cancel := downloadInput(t, custom.MapdInputType_cancelDownload, "")
	workerRunning := true
	defer func() {
		close(releases)
		if workerRunning {
			waitDownloadProgress(t, &settings)
		}
	}()
	settings.Handle(input)
	waitDownloadRequest(t, requests)
	<-settings.downloadProgress // Leave space for completion even on the unfixed baseline.
	if !settings.downloadActive {
		t.Fatal("download does not claim ownership until progress is consumed")
	}
	firstCancel := settings.cancelDownload
	settings.Handle(input)
	if settings.cancelDownload != firstCancel {
		t.Fatal("second command started another run")
	}
	settings.Handle(cancel)
	settings.Handle(cancel)
	releases <- struct{}{}
	terminal := waitDownloadProgress(t, &settings)
	workerRunning = false
	if terminal.Active || terminal.Canceled || terminal.DownloadedFiles != 1 {
		t.Fatalf("cancel must finish the in-flight final file: %+v", terminal)
	}
	if settings.downloadActive {
		t.Fatal("completed run still owns the downloader")
	}
	settings.Handle(cancel)
	settings.Handle(input)
	workerRunning = true
	waitDownloadRequest(t, requests)
	<-settings.downloadProgress
	if settings.cancelDownload == firstCancel || len(settings.cancelDownload) != 0 {
		t.Fatal("new run inherited the previous run's late cancellation")
	}
	releases <- struct{}{}
	terminal = waitDownloadProgress(t, &settings)
	workerRunning = false
	if terminal.Active || terminal.Canceled || terminal.DownloadedFiles != 1 || len(requests) != 0 {
		t.Fatalf("second run did not finish independently: %+v, extra requests %d", terminal, len(requests))
	}
}

func downloadInput(t *testing.T, inputType custom.MapdInputType, path string) custom.MapdIn {
	t.Helper()
	_, segment, err := capnp.NewMessage(capnp.SingleSegment(nil))
	if err != nil {
		t.Fatal(err)
	}
	input, err := custom.NewRootMapdIn(segment)
	if err != nil {
		t.Fatal(err)
	}
	input.SetType(inputType)
	if err := input.SetStr(path); err != nil {
		t.Fatal(err)
	}
	return input
}

func waitDownloadRequest(t *testing.T, requests <-chan struct{}) {
	t.Helper()
	select {
	case <-requests:
	case <-time.After(5 * time.Second):
		t.Fatal("download did not reach the controlled HTTP transport")
	}
}

func waitDownloadProgress(t *testing.T, settings *MapdSettings) DownloadProgress {
	t.Helper()
	deadline := time.Now().Add(5 * time.Second)
	for time.Now().Before(deadline) {
		if progress, ok := settings.GetDownloadProgress(); ok {
			return progress
		}
		time.Sleep(time.Millisecond)
	}
	t.Fatal("download did not publish completion")
	return DownloadProgress{}
}