bug(builder): avoid simultaneous git push

Author: aledbf

pkg/sshd/lock.go

@@ -0,0 +1,68 @@
+package sshd
+
+import (
+	"fmt"
+	"sync"
+	"time"
+)
+
+// RepositoryLock interface that allows the creation of a lock associated
+// with a repository name to avoid simultaneous git operations
+type RepositoryLock interface {
+	// Lock aquires a lock for a repository. In the case the repository is already locked
+	// it waits until a timeout to get the lock. If it was not possible to get the
+	// lock after the timeout an error is returned
+	Lock(repoName string, timeout time.Duration) error
+	// Unlock releases the lock for a repository or returns an error if the specified
+	// name doesn't exist. In the case the repository is already locked it waits until
+	// a timeout to get the lock. If it was not possible to get the lock after the timeout
+	// an error is returned
+	Unlock(repoName string, timeout time.Duration) error
+}
+
+// NewInMemoryRepositoryLock returns a new instance of a RepositoryLock
+func NewInMemoryRepositoryLock() RepositoryLock {
+	return &inMemoryRepoLock{
+		mutex:   &sync.RWMutex{},
+		dataMap: make(map[string]bool),
+	}
+}
+
+type inMemoryRepoLock struct {
+	mutex   *sync.RWMutex
+	dataMap map[string]bool
+}
+
+// Lock aquires a lock associated with the specified name.
+// This implementation do not uses the timeout
+func (rl *inMemoryRepoLock) Lock(repoName string, timeout time.Duration) error {
+	rl.mutex.Lock()
+	defer rl.mutex.Unlock()
+
+	_, exists := rl.dataMap[repoName]
+	if !exists {
+		rl.dataMap[repoName] = true
+		return nil
+	}
+
+	return fmt.Errorf("repository %q already locked", repoName)
+}
+
+// Unlock releases the lock for a repository or returns
+// an error if the specified name doesn't exist.
+// This implementation do not uses the timeout
+func (rl *inMemoryRepoLock) Unlock(repoName string, timeout time.Duration) error {
+	rl.mutex.Lock()
+	defer rl.mutex.Unlock()
+
+	locked, exists := rl.dataMap[repoName]
+	if !exists {
+		return fmt.Errorf("repository %q not found", repoName)
+	}
+
+	if locked {
+		delete(rl.dataMap, repoName)
+	}
+
+	return nil
+}

pkg/sshd/lock_test.go

@@ -0,0 +1,85 @@
+package sshd
+
+import (
+	"testing"
+	"time"
+)
+
+const (
+	callbackTimeout = 1 * time.Second
+)
+
+func TestSingleLock(t *testing.T) {
+	rl := NewInMemoryRepositoryLock()
+	key := "fakeid"
+	callbackCh := make(chan interface{})
+	go lockAndCallback(rl, key, callbackCh)
+	verifyCallbackHappens(t, callbackCh)
+}
+
+func TestSingleLockUnlock(t *testing.T) {
+	rl := NewInMemoryRepositoryLock()
+	key := "fakeid"
+	callbackCh := make(chan interface{})
+	go lockAndCallback(rl, key, callbackCh)
+	verifyCallbackHappens(t, callbackCh)
+	err := rl.Unlock(key, time.Duration(0))
+	if err != nil {
+		t.Fatalf("unexpected error %v", err)
+	}
+}
+
+func TestInvalidUnlock(t *testing.T) {
+	rl := NewInMemoryRepositoryLock()
+	key := "fakeid"
+	err := rl.Unlock(key, time.Duration(0))
+	if err == nil {
+		t.Fatalf("expected error but returned nil", err)
+	}
+}
+
+func TestDoubleLockUnlock(t *testing.T) {
+	rl := NewInMemoryRepositoryLock()
+	key := "fakeid"
+	callbackCh1stLock := make(chan interface{})
+	callbackCh2ndLock := make(chan interface{})
+
+	go lockAndCallback(rl, key, callbackCh1stLock)
+	verifyCallbackHappens(t, callbackCh1stLock)
+	go lockAndCallback(rl, key, callbackCh2ndLock)
+	verifyCallbackDoesntHappens(t, callbackCh2ndLock)
+	err := rl.Unlock(key, time.Duration(0))
+	if err != nil {
+		t.Fatalf("unexpected error %v", err)
+	}
+	err = rl.Unlock(key, time.Duration(0))
+	if err == nil {
+		t.Fatalf("expected error but returned nil")
+	}
+}
+
+func lockAndCallback(rl RepositoryLock, id string, callbackCh chan<- interface{}) {
+	if err := rl.Lock(id, time.Duration(0)); err == nil {
+		callbackCh <- true
+	}
+}
+
+func verifyCallbackHappens(t *testing.T, callbackCh <-chan interface{}) bool {
+	select {
+	case <-callbackCh:
+		return true
+	case <-time.After(callbackTimeout):
+		t.Fatalf("Timed out waiting for callback.")
+		return false
+	}
+}
+
+func verifyCallbackDoesntHappens(t *testing.T, callbackCh <-chan interface{}) bool {
+	select {
+	case <-callbackCh:
+		t.Fatalf("Unexpected callback.")
+		return false
+	case <-time.After(callbackTimeout):
+		return true
+	}
+}

pkg/sshd/server.go

@@ -9,10 +9,12 @@ package sshd
 
 import (
 	"fmt"
+	"io"
 	"net"
 	"strings"
 	"sync"
 	"text/template"
+	"time"
 
 	"github.com/Masterminds/cookoo"
 	"github.com/Masterminds/cookoo/log"
@@ -27,6 +29,12 @@ const (
 	Address string = "ssh.Address"
 	// ServerConfig is the context key for ServerConfig object.
 	ServerConfig string = "ssh.ServerConfig"
+
+	multiplePush string = "Another git push is ongoing"
+)
+
+var (
+	buildingRepos = NewInMemoryRepositoryLock()
 )
 
 // Serve starts a native SSH server.
@@ -215,6 +223,19 @@ func (s *server) answer(channel ssh.Channel, requests <-chan *ssh.Request, sshCo
 					req.Reply(ok, nil)
 					break
 				}
+
+				repoName := parts[1]
+				if err := buildingRepos.Lock(repoName, time.Duration(0)); err != nil {
+					log.Errf(s.c, multiplePush)
+					// The error must be in git format
+					if err := gitPktLine(channel, fmt.Sprintf("ERR %v\n", multiplePush)); err != nil {
+						log.Errf(s.c, "Failed to write to channel: %s", err)
+					}
+					sendExitStatus(1, channel)
+					req.Reply(false, nil)
+					return nil
+				}
+
 				req.Reply(true, nil) // We processed. Yay.
 
 				cxt.Put("channel", channel)
@@ -223,12 +244,14 @@ func (s *server) answer(channel ssh.Channel, requests <-chan *ssh.Request, sshCo
 				cxt.Put("repository", parts[1])
 				sshGitReceive := cxt.Get("route.sshd.sshGitReceive", "sshGitReceive").(string)
 				err := router.HandleRequest(sshGitReceive, cxt, true)
+				buildingRepos.Unlock(repoName, time.Duration(0))
 				var xs uint32
 				if err != nil {
 					log.Errf(s.c, "Failed git receive: %v", err)
 					xs = 1
 				}
 				sendExitStatus(xs, channel)
+
 				return nil
 			default:
 				log.Warnf(s.c, "Illegal command is '%s'\n", clean)
@@ -296,3 +319,10 @@ func Ping(c cookoo.Context, p *cookoo.Params) (interface{}, cookoo.Interrupt) {
 	req.Reply(true, nil)
 	return nil, nil
 }
+
+// gitPktLine writes a line following the git protocol
+// https://github.com/git/git/blob/master/Documentation/technical/pack-protocol.txt
+func gitPktLine(w io.Writer, s string) error {
+	_, err := fmt.Fprintf(w, "%04x%s", len(s)+4, s)
+	return err
+}