git-annex: start archive downloads immediately

Author: matrss

modules/annex/utils.go

@@ -0,0 +1,41 @@
+// Copyright 2023 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package annex
+
+import (
+	"errors"
+	"io"
+	"io/fs"
+	"math/rand"
+	"os"
+	"time"
+)
+
+// UntilFileDeletedReader is a io.Reader that reads from reader until the file at path
+// is deleted and reader does not return any new bytes.
+type UntilFileDeletedReader struct {
+	path   string
+	reader io.Reader
+}
+
+func NewUntilFileDeletedReader(path string, reader io.Reader) UntilFileDeletedReader {
+	return UntilFileDeletedReader{path, reader}
+}
+
+func (r UntilFileDeletedReader) Read(buf []byte) (n int, err error) {
+	n, err = r.reader.Read(buf)
+	if err != io.EOF {
+		return n, err
+	}
+	if n != 0 {
+		return n, nil
+	}
+	_, err = os.Stat(r.path)
+	if errors.Is(err, fs.ErrNotExist) {
+		return 0, io.EOF
+	}
+	// Avoid hammering Read when there is no data, sleep for 0 to 1 seconds
+	time.Sleep(time.Duration(rand.Intn(int(time.Second))))
+	return 0, nil
+}

routers/api/v1/repo/file.go

@@ -10,15 +10,19 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"math/rand"
 	"net/http"
+	"os"
 	"path"
+	"strconv"
 	"strings"
 	"time"
 
 	"code.gitea.io/gitea/models"
 	git_model "code.gitea.io/gitea/models/git"
 	repo_model "code.gitea.io/gitea/models/repo"
 	"code.gitea.io/gitea/models/unit"
+	"code.gitea.io/gitea/modules/annex"
 	"code.gitea.io/gitea/modules/context"
 	"code.gitea.io/gitea/modules/git"
 	"code.gitea.io/gitea/modules/httpcache"
@@ -307,13 +311,39 @@ func archiveDownload(ctx *context.APIContext) {
 		return
 	}
 
-	archiver, err := aReq.Await(ctx)
+	err = archiver_service.StartArchive(aReq)
 	if err != nil {
-		ctx.ServerError("archiver.Await", err)
+		ctx.ServerError("archiver_service.StartArchive", err)
 		return
 	}
 
-	download(ctx, aReq.GetArchiveName(), archiver)
+	for {
+		archiver, err := repo_model.GetRepoArchiver(ctx, aReq.RepoID, aReq.Type, aReq.CommitID)
+		if err != nil {
+			ctx.ServerError("repo_model.GetRepoArchiver", err)
+			return
+		}
+		if archiver != nil && archiver.Status == repo_model.ArchiverReady {
+			download(ctx, aReq.GetArchiveName(), archiver)
+			return
+		}
+		if archiver != nil && archiver.Status == repo_model.ArchiverGenerating {
+			filePath := os.TempDir() + "/" + strconv.FormatInt(archiver.RepoID, 10) + "-" + archiver.CommitID + "." + archiver.Type.String()
+			f, err := os.Open(filePath)
+			if err != nil {
+				// The archiver might have finished in-between repo_model.GetRepoArchiver and os.Open, retry
+				continue
+			}
+			defer f.Close()
+			r := annex.NewUntilFileDeletedReader(filePath, f)
+			downloadName := ctx.Repo.Repository.Name + "-" + aReq.GetArchiveName()
+			// TODO: figure out how to serve the file with an approximate size of the resulting archive
+			common.ServeContentByReader(ctx.Base, downloadName, -1, r)
+			return
+		}
+		// archiver was nil, retry after 0 to 1 seconds
+		time.Sleep(time.Duration(rand.Intn(int(time.Second))))
+	}
 }
 
 func download(ctx *context.APIContext, archiveName string, archiver *repo_model.RepoArchiver) {

routers/web/repo/repo.go

@@ -7,9 +7,13 @@ package repo
 import (
 	"errors"
 	"fmt"
+	"math/rand"
 	"net/http"
+	"os"
 	"slices"
+	"strconv"
 	"strings"
+	"time"
 
 	"code.gitea.io/gitea/models"
 	"code.gitea.io/gitea/models/db"
@@ -19,6 +23,7 @@ import (
 	repo_model "code.gitea.io/gitea/models/repo"
 	"code.gitea.io/gitea/models/unit"
 	user_model "code.gitea.io/gitea/models/user"
+	"code.gitea.io/gitea/modules/annex"
 	"code.gitea.io/gitea/modules/base"
 	"code.gitea.io/gitea/modules/cache"
 	"code.gitea.io/gitea/modules/context"
@@ -30,6 +35,7 @@ import (
 	api "code.gitea.io/gitea/modules/structs"
 	"code.gitea.io/gitea/modules/util"
 	"code.gitea.io/gitea/modules/web"
+	"code.gitea.io/gitea/routers/common"
 	"code.gitea.io/gitea/services/convert"
 	"code.gitea.io/gitea/services/forms"
 	repo_service "code.gitea.io/gitea/services/repository"
@@ -429,13 +435,39 @@ func Download(ctx *context.Context) {
 		return
 	}
 
-	archiver, err := aReq.Await(ctx)
+	err = archiver_service.StartArchive(aReq)
 	if err != nil {
-		ctx.ServerError("archiver.Await", err)
+		ctx.ServerError("archiver_service.StartArchive", err)
 		return
 	}
 
-	download(ctx, aReq.GetArchiveName(), archiver)
+	for {
+		archiver, err := repo_model.GetRepoArchiver(ctx, aReq.RepoID, aReq.Type, aReq.CommitID)
+		if err != nil {
+			ctx.ServerError("repo_model.GetRepoArchiver", err)
+			return
+		}
+		if archiver != nil && archiver.Status == repo_model.ArchiverReady {
+			download(ctx, aReq.GetArchiveName(), archiver)
+			return
+		}
+		if archiver != nil && archiver.Status == repo_model.ArchiverGenerating {
+			filePath := os.TempDir() + "/" + strconv.FormatInt(archiver.RepoID, 10) + "-" + archiver.CommitID + "." + archiver.Type.String()
+			f, err := os.Open(filePath)
+			if err != nil {
+				// The archiver might have finished in-between repo_model.GetRepoArchiver and os.Open, retry
+				continue
+			}
+			defer f.Close()
+			r := annex.NewUntilFileDeletedReader(filePath, f)
+			downloadName := ctx.Repo.Repository.Name + "-" + aReq.GetArchiveName()
+			// TODO: figure out how to serve the file with an approximate size of the resulting archive
+			common.ServeContentByReader(ctx.Base, downloadName, -1, r)
+			return
+		}
+		// archiver was nil, retry after 0 to 1 seconds
+		time.Sleep(time.Duration(rand.Intn(int(time.Second))))
+	}
 }
 
 func download(ctx *context.Context, archiveName string, archiver *repo_model.RepoArchiver) {
@@ -465,43 +497,6 @@ func download(ctx *context.Context, archiveName string, archiver *repo_model.Rep
 	})
 }
 
-// InitiateDownload will enqueue an archival request, as needed.  It may submit
-// a request that's already in-progress, but the archiver service will just
-// kind of drop it on the floor if this is the case.
-func InitiateDownload(ctx *context.Context) {
-	uri := ctx.Params("*")
-	aReq, err := archiver_service.NewRequest(ctx.Repo.Repository.ID, ctx.Repo.GitRepo, uri)
-	if err != nil {
-		ctx.ServerError("archiver_service.NewRequest", err)
-		return
-	}
-	if aReq == nil {
-		ctx.Error(http.StatusNotFound)
-		return
-	}
-
-	archiver, err := repo_model.GetRepoArchiver(ctx, aReq.RepoID, aReq.Type, aReq.CommitID)
-	if err != nil {
-		ctx.ServerError("archiver_service.StartArchive", err)
-		return
-	}
-	if archiver == nil || archiver.Status != repo_model.ArchiverReady {
-		if err := archiver_service.StartArchive(aReq); err != nil {
-			ctx.ServerError("archiver_service.StartArchive", err)
-			return
-		}
-	}
-
-	var completed bool
-	if archiver != nil && archiver.Status == repo_model.ArchiverReady {
-		completed = true
-	}
-
-	ctx.JSON(http.StatusOK, map[string]any{
-		"complete": completed,
-	})
-}
-
 // SearchRepo repositories via options
 func SearchRepo(ctx *context.Context) {
 	opts := &repo_model.SearchRepoOptions{

routers/web/web.go

@@ -1368,7 +1368,6 @@ func registerRoutes(m *web.Route) {
 
 		m.Group("/archive", func() {
 			m.Get("/*", repo.Download)
-			m.Post("/*", repo.InitiateDownload)
 		}, repo.MustBeNotEmpty, dlSourceEnabled, reqRepoCodeReader)
 
 		m.Group("/branches", func() {

services/repository/archiver/archiver.go

@@ -10,6 +10,7 @@ import (
 	"io"
 	"os"
 	"regexp"
+	"strconv"
 	"strings"
 	"time"
 
@@ -130,49 +131,6 @@ func (aReq *ArchiveRequest) GetArchiveName() string {
 	return strings.ReplaceAll(aReq.refName, "/", "-") + "." + aReq.Type.String()
 }
 
-// Await awaits the completion of an ArchiveRequest. If the archive has
-// already been prepared the method returns immediately. Otherwise an archiver
-// process will be started and its completion awaited. On success the returned
-// RepoArchiver may be used to download the archive. Note that even if the
-// context is cancelled/times out a started archiver will still continue to run
-// in the background.
-func (aReq *ArchiveRequest) Await(ctx context.Context) (*repo_model.RepoArchiver, error) {
-	archiver, err := repo_model.GetRepoArchiver(ctx, aReq.RepoID, aReq.Type, aReq.CommitID)
-	if err != nil {
-		return nil, fmt.Errorf("models.GetRepoArchiver: %w", err)
-	}
-
-	if archiver != nil && archiver.Status == repo_model.ArchiverReady {
-		// Archive already generated, we're done.
-		return archiver, nil
-	}
-
-	if err := StartArchive(aReq); err != nil {
-		return nil, fmt.Errorf("archiver.StartArchive: %w", err)
-	}
-
-	poll := time.NewTicker(time.Second * 1)
-	defer poll.Stop()
-
-	for {
-		select {
-		case <-graceful.GetManager().HammerContext().Done():
-			// System stopped.
-			return nil, graceful.GetManager().HammerContext().Err()
-		case <-ctx.Done():
-			return nil, ctx.Err()
-		case <-poll.C:
-			archiver, err = repo_model.GetRepoArchiver(ctx, aReq.RepoID, aReq.Type, aReq.CommitID)
-			if err != nil {
-				return nil, fmt.Errorf("repo_model.GetRepoArchiver: %w", err)
-			}
-			if archiver != nil && archiver.Status == repo_model.ArchiverReady {
-				return archiver, nil
-			}
-		}
-	}
-}
-
 func doArchive(ctx context.Context, r *ArchiveRequest) (*repo_model.RepoArchiver, error) {
 	txCtx, committer, err := db.TxContext(ctx)
 	if err != nil {
@@ -285,7 +243,15 @@ func doArchive(ctx context.Context, r *ArchiveRequest) (*repo_model.RepoArchiver
 	}
 	committer.Close()
 
-	if _, err := storage.RepoArchives.Save(rPath, rd, -1); err != nil {
+	f, err := os.Create(os.TempDir() + "/" + strconv.FormatInt(archiver.RepoID, 10) + "-" + archiver.CommitID + "." + archiver.Type.String())
+	if err != nil {
+		return nil, err
+	}
+	defer f.Close()
+	defer os.Remove(f.Name())
+
+	trd := io.TeeReader(rd, f)
+	if _, err := storage.RepoArchives.Save(rPath, trd, -1); err != nil {
 		return nil, fmt.Errorf("unable to write archive: %w", err)
 	}
 

web_src/js/features/repo-common.js

@@ -1,49 +1,6 @@
 import $ from 'jquery';
 import {hideElem, showElem} from '../utils/dom.js';
 
-const {csrfToken} = window.config;
-
-function getArchive($target, url, first) {
-  $.ajax({
-    url,
-    type: 'POST',
-    data: {
-      _csrf: csrfToken,
-    },
-    complete(xhr) {
-      if (xhr.status === 200) {
-        if (!xhr.responseJSON) {
-          // XXX Shouldn't happen?
-          $target.closest('.dropdown').children('i').removeClass('loading');
-          return;
-        }
-
-        if (!xhr.responseJSON.complete) {
-          $target.closest('.dropdown').children('i').addClass('loading');
-          // Wait for only three quarters of a second initially, in case it's
-          // quickly archived.
-          setTimeout(() => {
-            getArchive($target, url, false);
-          }, first ? 750 : 2000);
-        } else {
-          // We don't need to continue checking.
-          $target.closest('.dropdown').children('i').removeClass('loading');
-          window.location.href = url;
-        }
-      }
-    },
-  });
-}
-
-export function initRepoArchiveLinks() {
-  $('.archive-link').on('click', function (event) {
-    event.preventDefault();
-    const url = $(this).attr('href');
-    if (!url) return;
-    getArchive($(event.target), url, true);
-  });
-}
-
 export function initRepoCloneLink() {
   const $repoCloneSsh = $('#repo-clone-ssh');
   const $repoCloneHttps = $('#repo-clone-https');

web_src/js/index.js

@@ -56,7 +56,6 @@ import {initRepoTemplateSearch} from './features/repo-template.js';
 import {initRepoCodeView} from './features/repo-code.js';
 import {initSshKeyFormParser} from './features/sshkey-helper.js';
 import {initUserSettings} from './features/user-settings.js';
-import {initRepoArchiveLinks} from './features/repo-common.js';
 import {initRepoMigrationStatusChecker} from './features/repo-migrate.js';
 import {
   initRepoSettingGitHook,
@@ -139,7 +138,6 @@ onDomReady(() => {
   initOrgTeamSettings();
 
   initRepoActivityTopAuthorsChart();
-  initRepoArchiveLinks();
   initRepoBranchButton();
   initRepoCodeView();
   initRepoCommentForm();