From c92bda11db7ba1936d58611182b5ecf0ec3efdba Mon Sep 17 00:00:00 2001 From: James <91348155+FrogAi@users.noreply.github.com> Date: Sat, 8 Aug 2026 18:23:08 -0700 Subject: [PATCH 1/2] Install map archives transactionally --- settings/download.go | 279 +++++++++++++++++++++++++++++++++---------- 1 file changed, 217 insertions(+), 62 deletions(-) diff --git a/settings/download.go b/settings/download.go index 010c34c..6544c20 100644 --- a/settings/download.go +++ b/settings/download.go @@ -147,6 +147,216 @@ func Download(paths string, progressChan chan DownloadProgress, cancelChan chan } } +type archiveFile interface { + io.Writer + Sync() error + Close() error +} + +type archiveInstallOps struct { + mkdirTemp func(string, string) (string, error) + mkdirAll func(string, os.FileMode) error + openFile func(string, int, os.FileMode) (archiveFile, error) + rename func(string, string) error + removeAll func(string) error + stat func(string) (os.FileInfo, error) +} + +var defaultArchiveInstallOps = archiveInstallOps{ + mkdirTemp: os.MkdirTemp, + mkdirAll: os.MkdirAll, + openFile: func(name string, flag int, perm os.FileMode) (archiveFile, error) { + return os.OpenFile(name, flag, perm) + }, + rename: os.Rename, + removeAll: os.RemoveAll, + stat: os.Stat, +} + +func cleanArchivePath(name string) (string, error) { + clean := filepath.Clean(name) + if name == "" || clean == "." || filepath.IsAbs(clean) || clean == ".." || strings.HasPrefix(clean, ".."+string(os.PathSeparator)) { + return "", errors.Errorf("invalid archive path: %q", name) + } + return clean, nil +} + +func writeArchiveFile(file archiveFile, reader io.Reader, expectedSize int64) error { + written, err := io.Copy(file, reader) + if err != nil { + file.Close() + return errors.Wrap(err, "could not write staged archive file") + } + if written != expectedSize { + file.Close() + return errors.Errorf("staged archive file size mismatch: wrote %d bytes, expected %d", written, expectedSize) + } + if err := file.Sync(); err != nil { + file.Close() + return errors.Wrap(err, "could not fsync staged archive file") + } + if err := file.Close(); err != nil { + return errors.Wrap(err, "could not close staged archive file") + } + return nil +} + +func extractArchiveToStage(archivePath string, stageRoot string, expectedRoot string, ops archiveInstallOps) error { + file, err := os.Open(archivePath) + if err != nil { + return errors.Wrap(err, "could not open downloaded file") + } + defer file.Close() + + reader, err := gzip.NewReader(file) + if err != nil { + return errors.Wrap(err, "could not parse gzip downloaded file") + } + defer reader.Close() + + tr := tar.NewReader(reader) + seen := make(map[string]struct{}) + regularFiles := 0 + for { + header, err := tr.Next() + if err == io.EOF { + break + } + if err != nil { + return errors.Wrap(err, "could not read downloaded tar archive") + } + + // if the header is nil, just skip it (not sure how this happens) + if header == nil { + continue + } + + name, err := cleanArchivePath(header.Name) + if err != nil { + return err + } + if name != expectedRoot && !strings.HasPrefix(name, expectedRoot+string(os.PathSeparator)) { + return errors.Errorf("archive entry %q is outside expected root %q", header.Name, expectedRoot) + } + if _, exists := seen[name]; exists { + return errors.Errorf("duplicate archive entry: %q", header.Name) + } + seen[name] = struct{}{} + + // the target location where the dir/file should be created + target := filepath.Join(stageRoot, name) + // check the file type + switch header.Typeflag { + + // if its a dir and it doesn't exist create it + case tar.TypeDir: + if _, err := ops.stat(target); err != nil { + if !os.IsNotExist(err) { + return errors.Wrap(err, "could not inspect staged archive directory") + } + if err := ops.mkdirAll(target, os.FileMode(header.Mode).Perm()); err != nil { + return errors.Wrap(err, "could not create staged archive directory") + } + } + + // if it's a file create it + case tar.TypeReg, tar.TypeRegA: + if err := ops.mkdirAll(filepath.Dir(target), 0o755); err != nil { + return errors.Wrap(err, "could not create staged archive parent directory") + } + stagedFile, err := ops.openFile(target, os.O_CREATE|os.O_EXCL|os.O_WRONLY, os.FileMode(header.Mode).Perm()) + if err != nil { + return errors.Wrap(err, "could not open staged archive file") + } + if err := writeArchiveFile(stagedFile, tr, header.Size); err != nil { + return err + } + regularFiles++ + default: + return errors.Errorf("unsupported archive entry type %d for %q", header.Typeflag, header.Name) + } + } + + if _, err := io.Copy(io.Discard, reader); err != nil { + return errors.Wrap(err, "could not validate gzip trailer") + } + if regularFiles == 0 { + return errors.New("downloaded archive contains no regular files") + } + return nil +} + +func commitArchiveGroup(stageRoot string, basePath string, expectedRoot string, ops archiveInstallOps) (bool, error) { + stagedGroup := filepath.Join(stageRoot, expectedRoot) + stagedInfo, err := ops.stat(stagedGroup) + if err != nil { + return false, errors.Wrap(err, "could not inspect staged archive group") + } + if !stagedInfo.IsDir() { + return false, errors.Errorf("staged archive root is not a directory: %q", expectedRoot) + } + + liveGroup := filepath.Join(basePath, expectedRoot) + if err := ops.mkdirAll(filepath.Dir(liveGroup), 0o755); err != nil { + return false, errors.Wrap(err, "could not create live archive parent directory") + } + + backupGroup := filepath.Join(stageRoot, "backup") + hadLiveGroup := false + if _, err := ops.stat(liveGroup); err == nil { + if err := ops.rename(liveGroup, backupGroup); err != nil { + return false, errors.Wrap(err, "could not move live archive group to backup") + } + hadLiveGroup = true + } else if !os.IsNotExist(err) { + return false, errors.Wrap(err, "could not inspect live archive group") + } + + if err := ops.rename(stagedGroup, liveGroup); err != nil { + if hadLiveGroup { + if restoreErr := ops.rename(backupGroup, liveGroup); restoreErr != nil { + return true, errors.Wrapf(err, "could not install staged archive group and could not restore backup at %q: %v", backupGroup, restoreErr) + } + } + return false, errors.Wrap(err, "could not install staged archive group") + } + return false, nil +} + +func installArchiveWithOps(archivePath string, basePath string, expectedRoot string, ops archiveInstallOps) error { + cleanExpectedRoot, err := cleanArchivePath(expectedRoot) + if err != nil { + return err + } + if cleanExpectedRoot != expectedRoot { + return errors.Errorf("archive root is not canonical: %q", expectedRoot) + } + + stageRoot, err := ops.mkdirTemp(basePath, ".mapd-install-") + if err != nil { + return errors.Wrap(err, "could not create archive staging directory") + } + preserveStage := false + defer func() { + if preserveStage { + return + } + if err := ops.removeAll(stageRoot); err != nil { + slog.Warn("could not remove archive staging directory", "error", err, "directory", stageRoot) + } + }() + + if err := extractArchiveToStage(archivePath, stageRoot, expectedRoot, ops); err != nil { + return err + } + preserveStage, err = commitArchiveGroup(stageRoot, basePath, expectedRoot, ops) + return err +} + +func installArchive(archivePath string, basePath string, expectedRoot string) error { + return installArchiveWithOps(archivePath, basePath, expectedRoot, defaultArchiveInstallOps) +} + func adjustedBounds(bounds Bounds) (int, int, int, int) { minLat := int(math.Floor(bounds.MinLat/float64(GROUP_AREA_BOX_DEGREES))) * GROUP_AREA_BOX_DEGREES minLon := int(math.Floor(bounds.MinLon/float64(GROUP_AREA_BOX_DEGREES))) * GROUP_AREA_BOX_DEGREES @@ -194,71 +404,16 @@ func (d *download) downloadBounds(bounds Bounds, locationName string) (err error slog.Warn("failed to download file, continuing to next", "error", err, "url", url, "file", outputName) continue } - file, err := os.Open(outputName) - if err != nil { - slog.Warn("failed to open downloaded file", "error", err, "file", outputName) - } - reader, err := gzip.NewReader(file) - if err != nil { - slog.Warn("failed to parse gzip downloaded file", "error", err, "file", outputName) + installErr := installArchive(outputName, params.GetBaseOpPath(), strings.TrimSuffix(filename, ".tar.gz")) + if installErr != nil { + slog.Warn("failed to install downloaded archive", "error", installErr, "file", outputName) } - tr := tar.NewReader(reader) - for { - header, err := tr.Next() - if err != nil { - break - } - - // if the header is nil, just skip it (not sure how this happens) - if header == nil { - continue - } - // the target location where the dir/file should be created - target := filepath.Join(params.GetBaseOpPath(), header.Name) - // check the file type - switch header.Typeflag { - - // if its a dir and it doesn't exist create it - case tar.TypeDir: - if _, err := os.Stat(target); err != nil { - err := os.MkdirAll(target, 0o755) - if err != nil { - slog.Warn("could not create directory from downloaded gzip", "error", err, "file", outputName, "directory", target) - } - } - - // if it's a file create it - case tar.TypeReg: - f, err := os.OpenFile(target, os.O_CREATE|os.O_RDWR, os.FileMode(header.Mode)) - if err != nil { - slog.Warn("could not open file target from downloaded gzip", "error", err, "file", outputName, "targetFile", target) - } - - _, err = io.Copy(f, tr) - if err != nil { - slog.Warn("could not write data to file target from downloaded gzip", "error", err, "file", outputName, "targetFile", target) - } - - err = f.Sync() - if err != nil { - slog.Warn("could not fsync file target from downloaded gzip", "error", err, "file", outputName, "targetFile", target) - } - f.Close() - } - } - err = reader.Close() - if err != nil { - slog.Warn("could not close gzip reader", "error", err) - } - err = file.Close() - if err != nil { - slog.Warn("could not close downloaded file", "error", err) - } - - err = os.Remove(outputName) - if err != nil { + if err := os.Remove(outputName); err != nil { slog.Warn("could not delete downloaded gzip file", "error", err) } + if installErr != nil { + continue + } d.progress.DownloadedFiles++ d.progress.LocationDetails[locationName].DownloadedFiles++ From c64d04eb769fbce8fd167390ff94657d9064adff Mon Sep 17 00:00:00 2001 From: FrogAi <91348155+FrogAi@users.noreply.github.com> Date: Sun, 9 Aug 2026 07:42:49 -0700 Subject: [PATCH 2/2] Reclaim orphaned staging dirs and drop the archive install seam --- settings/download.go | 85 ++++++++++++++++++-------------------------- 1 file changed, 35 insertions(+), 50 deletions(-) diff --git a/settings/download.go b/settings/download.go index 6544c20..6a7f5bf 100644 --- a/settings/download.go +++ b/settings/download.go @@ -147,41 +147,15 @@ func Download(paths string, progressChan chan DownloadProgress, cancelChan chan } } -type archiveFile interface { - io.Writer - Sync() error - Close() error -} - -type archiveInstallOps struct { - mkdirTemp func(string, string) (string, error) - mkdirAll func(string, os.FileMode) error - openFile func(string, int, os.FileMode) (archiveFile, error) - rename func(string, string) error - removeAll func(string) error - stat func(string) (os.FileInfo, error) -} - -var defaultArchiveInstallOps = archiveInstallOps{ - mkdirTemp: os.MkdirTemp, - mkdirAll: os.MkdirAll, - openFile: func(name string, flag int, perm os.FileMode) (archiveFile, error) { - return os.OpenFile(name, flag, perm) - }, - rename: os.Rename, - removeAll: os.RemoveAll, - stat: os.Stat, -} - func cleanArchivePath(name string) (string, error) { clean := filepath.Clean(name) - if name == "" || clean == "." || filepath.IsAbs(clean) || clean == ".." || strings.HasPrefix(clean, ".."+string(os.PathSeparator)) { + if clean == "." || filepath.IsAbs(clean) || clean == ".." || strings.HasPrefix(clean, ".."+string(os.PathSeparator)) { return "", errors.Errorf("invalid archive path: %q", name) } return clean, nil } -func writeArchiveFile(file archiveFile, reader io.Reader, expectedSize int64) error { +func writeArchiveFile(file *os.File, reader io.Reader, expectedSize int64) error { written, err := io.Copy(file, reader) if err != nil { file.Close() @@ -201,7 +175,7 @@ func writeArchiveFile(file archiveFile, reader io.Reader, expectedSize int64) er return nil } -func extractArchiveToStage(archivePath string, stageRoot string, expectedRoot string, ops archiveInstallOps) error { +func extractArchiveToStage(archivePath string, stageRoot string, expectedRoot string) error { file, err := os.Open(archivePath) if err != nil { return errors.Wrap(err, "could not open downloaded file") @@ -250,21 +224,21 @@ func extractArchiveToStage(archivePath string, stageRoot string, expectedRoot st // if its a dir and it doesn't exist create it case tar.TypeDir: - if _, err := ops.stat(target); err != nil { + if _, err := os.Stat(target); err != nil { if !os.IsNotExist(err) { return errors.Wrap(err, "could not inspect staged archive directory") } - if err := ops.mkdirAll(target, os.FileMode(header.Mode).Perm()); err != nil { + if err := os.MkdirAll(target, 0o755); err != nil { return errors.Wrap(err, "could not create staged archive directory") } } // if it's a file create it - case tar.TypeReg, tar.TypeRegA: - if err := ops.mkdirAll(filepath.Dir(target), 0o755); err != nil { + case tar.TypeReg: + if err := os.MkdirAll(filepath.Dir(target), 0o755); err != nil { return errors.Wrap(err, "could not create staged archive parent directory") } - stagedFile, err := ops.openFile(target, os.O_CREATE|os.O_EXCL|os.O_WRONLY, os.FileMode(header.Mode).Perm()) + stagedFile, err := os.OpenFile(target, os.O_CREATE|os.O_EXCL|os.O_WRONLY, os.FileMode(header.Mode).Perm()) if err != nil { return errors.Wrap(err, "could not open staged archive file") } @@ -286,9 +260,9 @@ func extractArchiveToStage(archivePath string, stageRoot string, expectedRoot st return nil } -func commitArchiveGroup(stageRoot string, basePath string, expectedRoot string, ops archiveInstallOps) (bool, error) { +func commitArchiveGroup(stageRoot string, basePath string, expectedRoot string) (bool, error) { stagedGroup := filepath.Join(stageRoot, expectedRoot) - stagedInfo, err := ops.stat(stagedGroup) + stagedInfo, err := os.Stat(stagedGroup) if err != nil { return false, errors.Wrap(err, "could not inspect staged archive group") } @@ -297,14 +271,14 @@ func commitArchiveGroup(stageRoot string, basePath string, expectedRoot string, } liveGroup := filepath.Join(basePath, expectedRoot) - if err := ops.mkdirAll(filepath.Dir(liveGroup), 0o755); err != nil { + if err := os.MkdirAll(filepath.Dir(liveGroup), 0o755); err != nil { return false, errors.Wrap(err, "could not create live archive parent directory") } backupGroup := filepath.Join(stageRoot, "backup") hadLiveGroup := false - if _, err := ops.stat(liveGroup); err == nil { - if err := ops.rename(liveGroup, backupGroup); err != nil { + if _, err := os.Stat(liveGroup); err == nil { + if err := os.Rename(liveGroup, backupGroup); err != nil { return false, errors.Wrap(err, "could not move live archive group to backup") } hadLiveGroup = true @@ -312,9 +286,9 @@ func commitArchiveGroup(stageRoot string, basePath string, expectedRoot string, return false, errors.Wrap(err, "could not inspect live archive group") } - if err := ops.rename(stagedGroup, liveGroup); err != nil { + if err := os.Rename(stagedGroup, liveGroup); err != nil { if hadLiveGroup { - if restoreErr := ops.rename(backupGroup, liveGroup); restoreErr != nil { + if restoreErr := os.Rename(backupGroup, liveGroup); restoreErr != nil { return true, errors.Wrapf(err, "could not install staged archive group and could not restore backup at %q: %v", backupGroup, restoreErr) } } @@ -323,7 +297,20 @@ func commitArchiveGroup(stageRoot string, basePath string, expectedRoot string, return false, nil } -func installArchiveWithOps(archivePath string, basePath string, expectedRoot string, ops archiveInstallOps) error { +func removeOrphanedStagingDirs(basePath string) { + matches, err := filepath.Glob(filepath.Join(basePath, ".mapd-install-*")) + if err != nil { + slog.Warn("could not scan for orphaned archive staging directories", "error", err) + return + } + for _, match := range matches { + if err := os.RemoveAll(match); err != nil { + slog.Warn("could not remove orphaned archive staging directory", "error", err, "directory", match) + } + } +} + +func installArchive(archivePath string, basePath string, expectedRoot string) error { cleanExpectedRoot, err := cleanArchivePath(expectedRoot) if err != nil { return err @@ -332,7 +319,7 @@ func installArchiveWithOps(archivePath string, basePath string, expectedRoot str return errors.Errorf("archive root is not canonical: %q", expectedRoot) } - stageRoot, err := ops.mkdirTemp(basePath, ".mapd-install-") + stageRoot, err := os.MkdirTemp(basePath, ".mapd-install-") if err != nil { return errors.Wrap(err, "could not create archive staging directory") } @@ -341,22 +328,18 @@ func installArchiveWithOps(archivePath string, basePath string, expectedRoot str if preserveStage { return } - if err := ops.removeAll(stageRoot); err != nil { + if err := os.RemoveAll(stageRoot); err != nil { slog.Warn("could not remove archive staging directory", "error", err, "directory", stageRoot) } }() - if err := extractArchiveToStage(archivePath, stageRoot, expectedRoot, ops); err != nil { + if err := extractArchiveToStage(archivePath, stageRoot, expectedRoot); err != nil { return err } - preserveStage, err = commitArchiveGroup(stageRoot, basePath, expectedRoot, ops) + preserveStage, err = commitArchiveGroup(stageRoot, basePath, expectedRoot) return err } -func installArchive(archivePath string, basePath string, expectedRoot string) error { - return installArchiveWithOps(archivePath, basePath, expectedRoot, defaultArchiveInstallOps) -} - func adjustedBounds(bounds Bounds) (int, int, int, int) { minLat := int(math.Floor(bounds.MinLat/float64(GROUP_AREA_BOX_DEGREES))) * GROUP_AREA_BOX_DEGREES minLon := int(math.Floor(bounds.MinLon/float64(GROUP_AREA_BOX_DEGREES))) * GROUP_AREA_BOX_DEGREES @@ -375,6 +358,8 @@ func adjustedBounds(bounds Bounds) (int, int, int, int) { func (d *download) downloadBounds(bounds Bounds, locationName string) (err error, cancel bool) { slog.Info("Downloading Bounds", "min_lat", bounds.MinLat, "min_lon", bounds.MinLon, "max_lat", bounds.MaxLat, "max_lon", bounds.MaxLon) + removeOrphanedStagingDirs(params.GetBaseOpPath()) + // clip given bounds to file areas minLat, minLon, maxLat, maxLon := adjustedBounds(bounds) d.progress.LocationDetails[locationName].TotalFiles = countFilesForBounds(bounds)