Marking RemoteRepo as being updated, with worker PID, checking for locks. #45 #114

deb/remote.go

@@ -16,9 +16,16 @@ import (
 	"path/filepath"
 	"strconv"
 	"strings"
+	"syscall"
 	"time"
 )
 
+// RemoteRepo statuses
+const (
+	MirrorIdle = iota
+	MirrorUpdating
+)
+
 // RemoteRepo represents remote (fetchable) Debian repository.
 //
 // Repostitory could be filtered when fetching by components, architectures
@@ -49,8 +56,14 @@ type RemoteRepo struct {
 	Filter string
 	// FilterWithDeps to include dependencies from filter query
 	FilterWithDeps bool
+	// Status marks state of repository (being updated, no action)
+	Status int
+	// WorkerPID is PID of the process modifying the mirror (if any)
+	WorkerPID int
 	// "Snapshot" of current list of packages
 	packageRefs *PackageRefList
+	// Temporary list of package refs
+	tempPackageRefs *PackageRefList
 	// Parsed archived root
 	archiveRootURL *url.URL
 	// Current list of packages (filled while updating mirror)
@@ -142,6 +155,37 @@ func (repo *RemoteRepo) RefList() *PackageRefList {
 	return repo.packageRefs
 }
 
+// MarkAsUpdating puts current PID and sets status to updating
+func (repo *RemoteRepo) MarkAsUpdating() {
+	repo.Status = MirrorUpdating
+	repo.WorkerPID = os.Getpid()
+}
+
+// MarkAsIdle clears updating flag
+func (repo *RemoteRepo) MarkAsIdle() {
+	repo.Status = MirrorIdle
+	repo.WorkerPID = 0
+}
+
+// CheckLock returns error if mirror is being updated by another process
+func (repo *RemoteRepo) CheckLock() error {
+	if repo.Status == MirrorIdle || repo.WorkerPID == 0 {
+		return nil
+	}
+
+	p, err := os.FindProcess(repo.WorkerPID)
+	if err != nil {
+		return nil
+	}
+
+	err = p.Signal(syscall.Signal(0))
+	if err == nil {
+		return fmt.Errorf("mirror is locked by update operation, PID %d", repo.WorkerPID)
+	}
+
+	return nil
+}
+
 // ReleaseURL returns URL to Release* files in repo root
 func (repo *RemoteRepo) ReleaseURL(name string) *url.URL {
 	var path *url.URL
@@ -468,13 +512,19 @@ func (repo *RemoteRepo) BuildDownloadQueue(packagePool aptly.PackagePool) (queue
 		return
 	}
 
-	repo.packageRefs = NewPackageRefListFromPackageList(repo.packageList)
+	repo.tempPackageRefs = NewPackageRefListFromPackageList(repo.packageList)
 	// free up package list, we don't need it after this point
 	repo.packageList = nil
 
 	return
 }
 
+// FinalizeDownload swaps for final value of package refs
+func (repo *RemoteRepo) FinalizeDownload() {
+	repo.LastDownloadDate = time.Now()
+	repo.packageRefs = repo.tempPackageRefs
+}
+
 // Encode does msgpack encoding of RemoteRepo
 func (repo *RemoteRepo) Encode() []byte {
 	var buf bytes.Buffer

deb/remote_test.go

@@ -261,11 +261,13 @@ func (s *RemoteRepoSuite) TestDownload(c *C) {
 	c.Assert(s.downloader.Empty(), Equals, true)
 
 	queue, size, err := s.repo.BuildDownloadQueue(s.packagePool)
-	c.Assert(s.repo.packageRefs, NotNil)
 	c.Check(size, Equals, int64(3))
 	c.Check(queue, HasLen, 1)
 	c.Check(queue[0].RepoURI, Equals, "pool/main/a/amanda/amanda-client_3.3.1-3~bpo60+1_amd64.deb")
 
+	s.repo.FinalizeDownload()
+	c.Assert(s.repo.packageRefs, NotNil)
+
 	pkg, err := s.collectionFactory.PackageCollection().ByKey(s.repo.packageRefs.Refs[0])
 	c.Assert(err, IsNil)
 
@@ -304,6 +306,7 @@ func (s *RemoteRepoSuite) TestDownloadWithSources(c *C) {
 	c.Check(q[2], Equals, "pool/main/a/access-modifier-checker/access-modifier-checker_1.0.orig.tar.gz")
 	c.Check(q[0], Equals, "pool/main/a/access-modifier-checker/access-modifier-checker_1.0-4.debian.tar.gz")
 
+	s.repo.FinalizeDownload()
 	c.Assert(s.repo.packageRefs, NotNil)
 
 	pkg, err := s.collectionFactory.PackageCollection().ByKey(s.repo.packageRefs.Refs[0])
@@ -334,6 +337,8 @@ func (s *RemoteRepoSuite) TestDownloadFlat(c *C) {
 	c.Check(size, Equals, int64(3))
 	c.Check(queue, HasLen, 1)
 	c.Check(queue[0].RepoURI, Equals, "pool/main/a/amanda/amanda-client_3.3.1-3~bpo60+1_amd64.deb")
+
+	s.flat.FinalizeDownload()
 	c.Assert(s.flat.packageRefs, NotNil)
 
 	pkg, err := s.collectionFactory.PackageCollection().ByKey(s.flat.packageRefs.Refs[0])
@@ -353,10 +358,6 @@ func (s *RemoteRepoSuite) TestDownloadWithSourcesFlat(c *C) {
 	downloader.ExpectError("http://repos.express42.com/virool/precise/Sources.bz2", errors.New("HTTP 404"))
 	downloader.ExpectError("http://repos.express42.com/virool/precise/Sources.gz", errors.New("HTTP 404"))
 	downloader.ExpectResponse("http://repos.express42.com/virool/precise/Sources", exampleSourcesFile)
-	// downloader.AnyExpectResponse("http://repos.express42.com/virool/precise/pool/main/a/amanda/amanda-client_3.3.1-3~bpo60+1_amd64.deb", "xyz")
-	// downloader.AnyExpectResponse("http://repos.express42.com/virool/precise/pool/main/a/access-modifier-checker/access-modifier-checker_1.0-4.dsc", "abc")
-	// downloader.AnyExpectResponse("http://repos.express42.com/virool/precise/pool/main/a/access-modifier-checker/access-modifier-checker_1.0.orig.tar.gz", "abcd")
-	// downloader.AnyExpectResponse("http://repos.express42.com/virool/precise/pool/main/a/access-modifier-checker/access-modifier-checker_1.0-4.debian.tar.gz", "abcde")
 
 	err := s.flat.Fetch(downloader, nil)
 	c.Assert(err, IsNil)
@@ -379,6 +380,7 @@ func (s *RemoteRepoSuite) TestDownloadWithSourcesFlat(c *C) {
 	c.Check(q[2], Equals, "pool/main/a/access-modifier-checker/access-modifier-checker_1.0.orig.tar.gz")
 	c.Check(q[0], Equals, "pool/main/a/access-modifier-checker/access-modifier-checker_1.0-4.debian.tar.gz")
 
+	s.flat.FinalizeDownload()
 	c.Assert(s.flat.packageRefs, NotNil)
 
 	pkg, err := s.collectionFactory.PackageCollection().ByKey(s.flat.packageRefs.Refs[0])