From c8af09f1294059803a8f1b3142fd0b7a40fc9aba Mon Sep 17 00:00:00 2001
From: bigeagle <justin.w.xd@gmail.com>
Date: Tue, 2 Aug 2016 21:39:16 +0800
Subject: [PATCH] fix(worker): fix log file leakage of two-stage-sync provider.
 closing #36

---
 worker/loglimit_hook.go            |  2 +-
 worker/runner.go                   | 18 +++++++++++++++---
 worker/two_stage_rsync_provider.go |  1 +
 3 files changed, 17 insertions(+), 4 deletions(-)

diff --git a/worker/loglimit_hook.go b/worker/loglimit_hook.go
index ad89a84..b3ddffb 100644
--- a/worker/loglimit_hook.go
+++ b/worker/loglimit_hook.go
@@ -79,7 +79,7 @@ func (l *logLimiter) preExec() error {
 
 	logLink := filepath.Join(logDir, "latest")
 
-	if _, err = os.Stat(logLink); err == nil {
+	if _, err = os.Lstat(logLink); err == nil {
 		os.Remove(logLink)
 	}
 	os.Symlink(logFileName, logLink)
diff --git a/worker/runner.go b/worker/runner.go
index 04fc1fb..ff0399a 100644
--- a/worker/runner.go
+++ b/worker/runner.go
@@ -5,6 +5,7 @@ import (
 	"os"
 	"os/exec"
 	"strings"
+	"sync"
 	"syscall"
 	"time"
 
@@ -17,12 +18,14 @@ import (
 var errProcessNotStarted = errors.New("Process Not Started")
 
 type cmdJob struct {
+	sync.Mutex
 	cmd        *exec.Cmd
 	workingDir string
 	env        map[string]string
 	logFile    *os.File
 	finished   chan empty
 	provider   mirrorProvider
+	retErr     error
 }
 
 func newCmdJob(provider mirrorProvider, cmdAndArgs []string, workingDir string, env map[string]string) *cmdJob {
@@ -69,9 +72,18 @@ func (c *cmdJob) Start() error {
 }
 
 func (c *cmdJob) Wait() error {
-	err := c.cmd.Wait()
-	close(c.finished)
-	return err
+	c.Lock()
+	defer c.Unlock()
+
+	select {
+	case <-c.finished:
+		return c.retErr
+	default:
+		err := c.cmd.Wait()
+		c.retErr = err
+		close(c.finished)
+		return err
+	}
 }
 
 func (c *cmdJob) SetLogFile(logFile *os.File) {
diff --git a/worker/two_stage_rsync_provider.go b/worker/two_stage_rsync_provider.go
index 21324ca..48e5125 100644
--- a/worker/two_stage_rsync_provider.go
+++ b/worker/two_stage_rsync_provider.go
@@ -108,6 +108,7 @@ func (p *twoStageRsyncProvider) Options(stage int) ([]string, error) {
 }
 
 func (p *twoStageRsyncProvider) Run() error {
+	defer p.Wait()
 
 	env := map[string]string{}
 	if p.username != "" {