Merge pull request #32 from laszlocph/multi-pipeline

Multi pipeline

.drone.yml

@@ -1,3 +1,7 @@
+clone:
+  git:
+    image: plugins/git:next
+
 workspace:
   base: /go
   path: src/github.com/laszlocph/drone-oss-08
@@ -109,7 +113,7 @@ pipeline:
     repo: laszlocloud/drone-oss-08-server
     dockerfile: Dockerfile.alpine
     secrets: [ docker_username, docker_password ]
-    tag: [ 0.8.96-alpine ]
+    tag: [ 0.8.96-multi-pipeline-alpine ]
     when:
       event: tag
 
@@ -118,7 +122,7 @@ pipeline:
     repo: laszlocloud/drone-oss-08-agent
     dockerfile: Dockerfile.agent.alpine
     secrets: [ docker_username, docker_password ]
-    tag: [ 0.8.96-alpine ]
+    tag: [ 0.8.96-multi-pipeline-alpine ]
     when:
       event: tag
 
@@ -126,7 +130,7 @@ pipeline:
     image: plugins/docker
     repo: laszlocloud/drone-oss-08-server
     secrets: [ docker_username, docker_password ]
-    tag: [ 0.8.96 ]
+    tag: [ 0.8.96-multi-pipeline ]
     when:
       event: tag
 
@@ -135,7 +139,7 @@ pipeline:
     repo: laszlocloud/drone-oss-08-agent
     dockerfile: Dockerfile.agent
     secrets: [ docker_username, docker_password ]
-    tag: [ 0.8.96 ]
+    tag: [ 0.8.96-multi-pipeline ]
     when:
       event: tag
 

.gitignore

@@ -8,4 +8,5 @@ release/
 cli/release/
 
 server/swagger/files/*.json
+server/swagger/swagger_gen.go
 .idea/

BUILDING

@@ -10,3 +10,20 @@
 
     go install github.com/laszlocph/drone-oss-08/cmd/drone-agent
     go install github.com/laszlocph/drone-oss-08/cmd/drone-server
+
+---
+
+0. To generate SQL files
+
+go get github.com/vektra/mockery/.../
+
+export download_url=$(curl -s https://api.github.com/repos/go-swagger/go-swagger/releases/latest | \
+  jq -r '.assets[] | select(.name | contains("'"$(uname | tr '[:upper:]' '[:lower:]')"'_amd64")) | .browser_download_url')
+curl -o swagger -L'#' "$download_url"
+chmod +x swagger
+sudo mv swagger /usr/local/bin
+
+go get github.com/laszlocph/togo
+
+go generate
+

cncd/logging/log_test.go

@@ -30,7 +30,7 @@ func TestLogging(t *testing.T) {
 		logger.Tail(ctx, testPath, func(entry ...*Entry) { wg.Done() })
 	}()
 
-	<-time.After(time.Millisecond)
+	<-time.After(500 * time.Millisecond)
 
 	wg.Add(4)
 	go func() {
@@ -45,7 +45,7 @@ func TestLogging(t *testing.T) {
 		logger.Tail(ctx, testPath, func(entry ...*Entry) { wg.Done() })
 	}()
 
-	<-time.After(time.Millisecond)
+	<-time.After(500 * time.Millisecond)
 
 	wg.Wait()
 	cancel()

cncd/pipeline/pipeline/frontend/metadata.go

@@ -222,3 +222,10 @@ func (m *Metadata) EnvironDrone() map[string]string {
 }
 
 var pullRegexp = regexp.MustCompile("\\d+")
+
+func (m *Metadata) SetPlatform(platform string) {
+	if platform == "" {
+		platform = "linux/amd64"
+	}
+	m.Sys.Arch = platform
+}

cncd/pipeline/pipeline/frontend/yaml/compiler/compiler.go

@@ -97,7 +97,7 @@ func (c *Compiler) Compile(conf *yaml.Config) *backend.Config {
 	}
 
 	// add default clone step
-	if c.local == false && len(conf.Clone.Containers) == 0 {
+	if c.local == false && len(conf.Clone.Containers) == 0 && !conf.SkipClone {
 		container := &yaml.Container{
 			Name:  "clone",
 			Image: "plugins/git:latest",
@@ -118,7 +118,7 @@ func (c *Compiler) Compile(conf *yaml.Config) *backend.Config {
 		stage.Steps = append(stage.Steps, step)
 
 		config.Stages = append(config.Stages, stage)
-	} else if c.local == false {
+	} else if c.local == false && !conf.SkipClone {
 		for i, container := range conf.Clone.Containers {
 			if !container.Constraints.Match(c.metadata) {
 				continue

cncd/pipeline/pipeline/frontend/yaml/config.go

@@ -22,6 +22,9 @@ type (
 		Networks  Networks
 		Volumes   Volumes
 		Labels    libcompose.SliceorMap
+		DependsOn []string `yaml:"depends_on,omitempty"`
+		RunsOn    []string `yaml:"runs_on,omitempty"`
+		SkipClone bool     `yaml:"skip_clone"`
 	}
 
 	// Workspace defines a pipeline workspace.

cncd/pipeline/pipeline/frontend/yaml/config_test.go

@@ -7,7 +7,7 @@ import (
 	"github.com/franela/goblin"
 )
 
-func xTestParse(t *testing.T) {
+func TestParse(t *testing.T) {
 	g := goblin.Goblin(t)
 
 	g.Describe("Parser", func() {
@@ -35,9 +35,14 @@ func xTestParse(t *testing.T) {
 				g.Assert(out.Pipeline.Containers[1].Commands).Equal(yaml.Stringorslice{"go build"})
 				g.Assert(out.Pipeline.Containers[2].Name).Equal("notify")
 				g.Assert(out.Pipeline.Containers[2].Image).Equal("slack")
-				g.Assert(out.Pipeline.Containers[2].NetworkMode).Equal("container:name")
+				// g.Assert(out.Pipeline.Containers[2].NetworkMode).Equal("container:name")
 				g.Assert(out.Labels["com.example.team"]).Equal("frontend")
 				g.Assert(out.Labels["com.example.type"]).Equal("build")
+				g.Assert(out.DependsOn[0]).Equal("lint")
+				g.Assert(out.DependsOn[1]).Equal("test")
+				g.Assert(out.RunsOn[0]).Equal("success")
+				g.Assert(out.RunsOn[1]).Equal("failure")
+				g.Assert(out.SkipClone).Equal(false)
 			})
 			// Check to make sure variable expansion works in yaml.MapSlice
 			// g.It("Should unmarshal variables", func() {
@@ -94,6 +99,12 @@ volumes:
 labels:
   com.example.type: "build"
   com.example.team: "frontend"
+depends_on:
+  - lint
+  - test
+runs_on:
+  - success
+  - failure
 `
 
 var sampleVarYaml = `

cncd/pipeline/pipeline/frontend/yaml/matrix/matrix.go

@@ -40,9 +40,8 @@ func Parse(data []byte) ([]Axis, error) {
 		return nil, err
 	}
 
-	// if not a matrix build return an array with just the single axis.
 	if len(matrix) == 0 {
-		return nil, nil
+		return []Axis{}, nil
 	}
 
 	return calc(matrix), nil

cncd/pipeline/pipeline/frontend/yaml/matrix/matrix_test.go

@@ -25,10 +25,10 @@ func TestMatrix(t *testing.T) {
 			g.Assert(len(set)).Equal(24)
 		})
 
-		g.It("Should return nil if no matrix", func() {
+		g.It("Should return empty array if no matrix", func() {
 			axis, err := ParseString("")
 			g.Assert(err == nil).IsTrue()
-			g.Assert(axis == nil).IsTrue()
+			g.Assert(len(axis) == 0).IsTrue()
 		})
 
 		g.It("Should return included axis", func() {

cncd/pubsub/pub_test.go

@@ -30,7 +30,7 @@ func TestPubsub(t *testing.T) {
 		broker.Subscribe(ctx, testTopic, func(message Message) { wg.Done() })
 	}()
 
-	<-time.After(time.Millisecond)
+	<-time.After(500 * time.Millisecond)
 
 	if _, ok := broker.(*publisher).topics[testTopic]; !ok {
 		t.Errorf("Expect topic registered with publisher")
@@ -86,7 +86,7 @@ func TestSubscriptionClosed(t *testing.T) {
 		wg.Done()
 	}()
 
-	<-time.After(time.Millisecond)
+	<-time.After(500 * time.Millisecond)
 
 	if _, ok := broker.(*publisher).topics[testTopic]; !ok {
 		t.Errorf("Expect topic registered with publisher")

cncd/queue/fifo.go

@@ -7,6 +7,8 @@ import (
 	"runtime"
 	"sync"
 	"time"
+
+	"github.com/Sirupsen/logrus"
 )
 
 type entry struct {
@@ -50,6 +52,17 @@ func (q *fifo) Push(c context.Context, task *Task) error {
 	return nil
 }
 
+// Push pushes an item to the tail of this queue.
+func (q *fifo) PushAtOnce(c context.Context, tasks []*Task) error {
+	q.Lock()
+	for _, task := range tasks {
+		q.pending.PushBack(task)
+	}
+	q.Unlock()
+	go q.process()
+	return nil
+}
+
 // Poll retrieves and removes the head of this queue.
 func (q *fifo) Poll(c context.Context, f Filter) (*Task, error) {
 	q.Lock()
@@ -82,11 +95,14 @@ func (q *fifo) Done(c context.Context, id string) error {
 // Error signals that the item is done executing with error.
 func (q *fifo) Error(c context.Context, id string, err error) error {
 	q.Lock()
-	state, ok := q.running[id]
+	taskEntry, ok := q.running[id]
 	if ok {
-		state.error = err
+		q.updateDepStatusInQueue(id, err == nil)
-		close(state.done)
+		taskEntry.error = err
+		close(taskEntry.done)
 		delete(q.running, id)
+	} else {
+		q.removeFromPending(id)
 	}
 	q.Unlock()
 	return nil
@@ -173,8 +189,44 @@ func (q *fifo) process() {
 	q.Lock()
 	defer q.Unlock()
 
-	// TODO(bradrydzewski) move this to a helper function
+	q.resubmitExpiredBuilds()
-	// push items to the front of the queue if the item expires.
+
+	for pending, worker := q.assignToWorker(); pending != nil && worker != nil; pending, worker = q.assignToWorker() {
+		task := pending.Value.(*Task)
+		delete(q.workers, worker)
+		q.pending.Remove(pending)
+		q.running[task.ID] = &entry{
+			item:     task,
+			done:     make(chan bool),
+			deadline: time.Now().Add(q.extension),
+		}
+		worker.channel <- task
+	}
+}
+
+func (q *fifo) assignToWorker() (*list.Element, *worker) {
+	var next *list.Element
+	for e := q.pending.Front(); e != nil; e = next {
+		next = e.Next()
+		task := e.Value.(*Task)
+		logrus.Debugf("queue: trying to assign task: %v with deps %v", task.ID, task.Dependencies)
+		if q.depsInQueue(task) {
+			logrus.Debugf("queue: skipping due to unmet dependencies %v", task.ID)
+			continue
+		}
+
+		for w := range q.workers {
+			if w.filter(task) {
+				logrus.Debugf("queue: assigned task: %v with deps %v", task.ID, task.Dependencies)
+				return e, w
+			}
+		}
+	}
+
+	return nil, nil
+}
+
+func (q *fifo) resubmitExpiredBuilds() {
 	for id, state := range q.running {
 		if time.Now().After(state.deadline) {
 			q.pending.PushFront(state.item)
@@ -182,26 +234,61 @@ func (q *fifo) process() {
 			close(state.done)
 		}
 	}
+}
 
+func (q *fifo) depsInQueue(task *Task) bool {
 	var next *list.Element
-loop:
 	for e := q.pending.Front(); e != nil; e = next {
 		next = e.Next()
-		item := e.Value.(*Task)
+		possibleDep, ok := e.Value.(*Task)
-		for w := range q.workers {
+		logrus.Debugf("queue: pending right now: %v", possibleDep.ID)
-			if w.filter(item) {
+		for _, dep := range task.Dependencies {
-				delete(q.workers, w)
+			if ok && possibleDep.ID == dep {
-				q.pending.Remove(e)
+				return true
+			}
+		}
+	}
+	for possibleDepID := range q.running {
+		logrus.Debugf("queue: running right now: %v", possibleDepID)
+		for _, dep := range task.Dependencies {
+			if possibleDepID == dep {
+				return true
+			}
+		}
+	}
+	return false
+}
 
-				q.running[item.ID] = &entry{
+func (q *fifo) updateDepStatusInQueue(taskID string, success bool) {
-					item:     item,
+	var next *list.Element
-					done:     make(chan bool),
+	for e := q.pending.Front(); e != nil; e = next {
-					deadline: time.Now().Add(q.extension),
+		next = e.Next()
-				}
+		pending, ok := e.Value.(*Task)
-
+		for _, dep := range pending.Dependencies {
-				w.channel <- item
+			if ok && taskID == dep {
-				break loop
+				pending.DepStatus[dep] = success
+			}
+		}
+	}
+	for _, running := range q.running {
+		for _, dep := range running.item.Dependencies {
+			if taskID == dep {
+				running.item.DepStatus[dep] = success
 			}
 		}
 	}
 }
+
+func (q *fifo) removeFromPending(taskID string) {
+	logrus.Debugf("queue: trying to remove %s", taskID)
+	var next *list.Element
+	for e := q.pending.Front(); e != nil; e = next {
+		next = e.Next()
+		task := e.Value.(*Task)
+		if task.ID == taskID {
+			logrus.Debugf("queue: %s is removed from pending", taskID)
+			q.pending.Remove(e)
+			return
+		}
+	}
+}

cncd/queue/fifo_test.go

@@ -2,6 +2,7 @@ package queue
 
 import (
 	"context"
+	"fmt"
 	"sync"
 	"testing"
 	"time"
@@ -117,3 +118,188 @@ func TestFifoEvict(t *testing.T) {
 		t.Errorf("expect not found error when evicting item not in queue, got %s", err)
 	}
 }
+
+func TestFifoDependencies(t *testing.T) {
+	task1 := &Task{
+		ID: "1",
+	}
+
+	task2 := &Task{
+		ID:           "2",
+		Dependencies: []string{"1"},
+		DepStatus:    make(map[string]bool),
+	}
+
+	q := New().(*fifo)
+	q.Push(noContext, task2)
+	q.Push(noContext, task1)
+
+	got, _ := q.Poll(noContext, func(*Task) bool { return true })
+	if got != task1 {
+		t.Errorf("expect task1 returned from queue as task2 depends on it")
+		return
+	}
+
+	q.Done(noContext, got.ID)
+
+	got, _ = q.Poll(noContext, func(*Task) bool { return true })
+	if got != task2 {
+		t.Errorf("expect task2 returned from queue")
+		return
+	}
+}
+
+func TestFifoErrors(t *testing.T) {
+	task1 := &Task{
+		ID: "1",
+	}
+
+	task2 := &Task{
+		ID:           "2",
+		Dependencies: []string{"1"},
+		DepStatus:    make(map[string]bool),
+	}
+
+	task3 := &Task{
+		ID:           "3",
+		Dependencies: []string{"1"},
+		DepStatus:    make(map[string]bool),
+		RunOn:        []string{"success", "failure"},
+	}
+
+	q := New().(*fifo)
+	q.Push(noContext, task2)
+	q.Push(noContext, task3)
+	q.Push(noContext, task1)
+
+	got, _ := q.Poll(noContext, func(*Task) bool { return true })
+	if got != task1 {
+		t.Errorf("expect task1 returned from queue as task2 depends on it")
+		return
+	}
+
+	q.Error(noContext, got.ID, fmt.Errorf("exitcode 1, there was an error"))
+
+	got, _ = q.Poll(noContext, func(*Task) bool { return true })
+	if got != task2 {
+		t.Errorf("expect task2 returned from queue")
+		return
+	}
+
+	if got.ShouldRun() {
+		t.Errorf("expect task2 should not run, since task1 failed")
+		return
+	}
+
+	got, _ = q.Poll(noContext, func(*Task) bool { return true })
+	if got != task3 {
+		t.Errorf("expect task3 returned from queue")
+		return
+	}
+
+	if !got.ShouldRun() {
+		t.Errorf("expect task3 should run, task1 failed, but task3 runs on failure too")
+		return
+	}
+}
+
+func TestFifoCancel(t *testing.T) {
+	task1 := &Task{
+		ID: "1",
+	}
+
+	task2 := &Task{
+		ID:           "2",
+		Dependencies: []string{"1"},
+		DepStatus:    make(map[string]bool),
+	}
+
+	task3 := &Task{
+		ID:           "3",
+		Dependencies: []string{"1"},
+		DepStatus:    make(map[string]bool),
+		RunOn:        []string{"success", "failure"},
+	}
+
+	q := New().(*fifo)
+	q.Push(noContext, task2)
+	q.Push(noContext, task3)
+	q.Push(noContext, task1)
+
+	_, _ = q.Poll(noContext, func(*Task) bool { return true })
+	q.Error(noContext, task1.ID, fmt.Errorf("cancelled"))
+	q.Error(noContext, task2.ID, fmt.Errorf("cancelled"))
+	q.Error(noContext, task3.ID, fmt.Errorf("cancelled"))
+
+	info := q.Info(noContext)
+	if len(info.Pending) != 0 {
+		t.Errorf("All pipelines should be cancelled")
+		return
+	}
+}
+
+func TestShouldRun(t *testing.T) {
+	task := &Task{
+		ID:           "2",
+		Dependencies: []string{"1"},
+		DepStatus: map[string]bool{
+			"1": true,
+		},
+		RunOn: []string{"failure"},
+	}
+	if task.ShouldRun() {
+		t.Errorf("expect task to not run, it runs on failure only")
+		return
+	}
+
+	task = &Task{
+		ID:           "2",
+		Dependencies: []string{"1"},
+		DepStatus: map[string]bool{
+			"1": true,
+		},
+		RunOn: []string{"failure", "success"},
+	}
+	if !task.ShouldRun() {
+		t.Errorf("expect task to run")
+		return
+	}
+
+	task = &Task{
+		ID:           "2",
+		Dependencies: []string{"1"},
+		DepStatus: map[string]bool{
+			"1": false,
+		},
+	}
+	if task.ShouldRun() {
+		t.Errorf("expect task to not run")
+		return
+	}
+
+	task = &Task{
+		ID:           "2",
+		Dependencies: []string{"1"},
+		DepStatus: map[string]bool{
+			"1": true,
+		},
+		RunOn: []string{"success"},
+	}
+	if !task.ShouldRun() {
+		t.Errorf("expect task to run")
+		return
+	}
+
+	task = &Task{
+		ID:           "2",
+		Dependencies: []string{"1"},
+		DepStatus: map[string]bool{
+			"1": false,
+		},
+		RunOn: []string{"failure"},
+	}
+	if !task.ShouldRun() {
+		t.Errorf("expect task to run")
+		return
+	}
+}

cncd/queue/queue.go

@@ -23,6 +23,64 @@ type Task struct {
 
 	// Labels represents the key-value pairs the entry is lebeled with.
 	Labels map[string]string `json:"labels,omitempty"`
+
+	// Task IDs this task depend
+	Dependencies []string
+
+	// If dep finished sucessfully
+	DepStatus map[string]bool
+
+	// RunOn failure or success
+	RunOn []string
+}
+
+// ShouldRun tells if a task should be run or skipped, based on dependencies
+func (t *Task) ShouldRun() bool {
+	if runsOnFailure(t.RunOn) && runsOnSuccess(t.RunOn) {
+		return true
+	}
+
+	if !runsOnFailure(t.RunOn) && runsOnSuccess(t.RunOn) {
+		for _, success := range t.DepStatus {
+			if !success {
+				return false
+			}
+		}
+		return true
+	}
+
+	if runsOnFailure(t.RunOn) && !runsOnSuccess(t.RunOn) {
+		for _, success := range t.DepStatus {
+			if success {
+				return false
+			}
+		}
+		return true
+	}
+
+	return false
+}
+
+func runsOnFailure(runsOn []string) bool {
+	for _, status := range runsOn {
+		if status == "failure" {
+			return true
+		}
+	}
+	return false
+}
+
+func runsOnSuccess(runsOn []string) bool {
+	if len(runsOn) == 0 {
+		return true
+	}
+
+	for _, status := range runsOn {
+		if status == "success" {
+			return true
+		}
+	}
+	return false
 }
 
 // InfoT provides runtime information.
@@ -44,9 +102,12 @@ type Filter func(*Task) bool
 // Queue defines a task queue for scheduling tasks among
 // a pool of workers.
 type Queue interface {
-	// Push pushes an task to the tail of this queue.
+	// Push pushes a task to the tail of this queue.
 	Push(c context.Context, task *Task) error
 
+	// Push pushes a task to the tail of this queue.
+	PushAtOnce(c context.Context, tasks []*Task) error
+
 	// Poll retrieves and removes a task head of this queue.
 	Poll(c context.Context, f Filter) (*Task, error)
 
@@ -68,46 +129,3 @@ type Queue interface {
 	// Info returns internal queue information.
 	Info(c context.Context) InfoT
 }
-
-// // global instance of the queue.
-// var global = New()
-//
-// // Set sets the global queue.
-// func Set(queue Queue) {
-// 	global = queue
-// }
-//
-// // Push pushes an task to the tail of the global queue.
-// func Push(c context.Context, task *Task) error {
-// 	return global.Push(c, task)
-// }
-//
-// // Poll retrieves and removes a task head of the global queue.
-// func Poll(c context.Context, f Filter) (*Task, error) {
-// 	return global.Poll(c, f)
-// }
-//
-// // Extend extends the deadline for a task.
-// func Extend(c context.Context, id string) error {
-// 	return global.Extend(c, id)
-// }
-//
-// // Done signals the task is complete.
-// func Done(c context.Context, id string) error {
-// 	return global.Done(c, id)
-// }
-//
-// // Error signals the task is complete with errors.
-// func Error(c context.Context, id string, err error) {
-// 	global.Error(c, id, err)
-// }
-//
-// // Wait waits until the task is complete.
-// func Wait(c context.Context, id string) error {
-// 	return global.Wait(c, id)
-// }
-//
-// // Info returns internal queue information.
-// func Info(c context.Context) InfoT {
-// 	return global.Info(c)
-// }

docker-compose.yml

@@ -28,4 +28,4 @@ services:
     environment:
       - DRONE_SERVER=drone-server:9000
       - DRONE_SECRET=${DRONE_SECRET}
-      - DRONE_MAX_PROCS=1
+      - DRONE_MAX_PROCS=2

model/config.go

@@ -16,10 +16,11 @@ package model
 
 // ConfigStore persists pipeline configuration to storage.
 type ConfigStore interface {
-	ConfigLoad(int64) (*Config, error)
+	ConfigsForBuild(buildID int64) ([]*Config, error)
-	ConfigFind(*Repo, string) (*Config, error)
+	ConfigFindIdentical(repoID int64, sha string) (*Config, error)
 	ConfigFindApproved(*Config) (bool, error)
 	ConfigCreate(*Config) error
+	BuildConfigCreate(*BuildConfig) error
 }
 
 // Config represents a pipeline configuration.
@@ -28,4 +29,11 @@ type Config struct {
 	RepoID int64  `json:"-"    meddler:"config_repo_id"`
 	Data   string `json:"data" meddler:"config_data"`
 	Hash   string `json:"hash" meddler:"config_hash"`
+	Name   string `json:"name" meddler:"config_name"`
+}
+
+// BuildConfig is the n:n relation between Build and Config
+type BuildConfig struct {
+	ConfigID int64 `json:"-"    meddler:"config_id"`
+	BuildID  int64 `json:"-"    meddler:"build_id"`
 }

model/proc.go

@@ -14,6 +14,8 @@
 
 package model
 
+import "fmt"
+
 // ProcStore persists process information to storage.
 type ProcStore interface {
 	ProcLoad(int64) (*Proc, error)
@@ -57,18 +59,24 @@ func (p *Proc) Failing() bool {
 
 // Tree creates a process tree from a flat process list.
 func Tree(procs []*Proc) []*Proc {
-	var (
+	var nodes []*Proc
-		nodes  []*Proc
-		parent *Proc
-	)
 	for _, proc := range procs {
 		if proc.PPID == 0 {
 			nodes = append(nodes, proc)
-			parent = proc
-			continue
 		} else {
+			parent, _ := findNode(nodes, proc.PPID)
 			parent.Children = append(parent.Children, proc)
 		}
 	}
 	return nodes
 }
+
+func findNode(nodes []*Proc, pid int) (*Proc, error) {
+	for _, node := range nodes {
+		if node.PID == pid {
+			return node, nil
+		}
+	}
+
+	return nil, fmt.Errorf("Corrupt proc structure")
+}

model/queue.go

@@ -23,9 +23,11 @@ import (
 
 // Task defines scheduled pipeline Task.
 type Task struct {
-	ID     string            `meddler:"task_id"`
+	ID           string            `meddler:"task_id"`
-	Data   []byte            `meddler:"task_data"`
+	Data         []byte            `meddler:"task_data"`
-	Labels map[string]string `meddler:"task_labels,json"`
+	Labels       map[string]string `meddler:"task_labels,json"`
+	Dependencies []string          `meddler:"task_dependencies,json"`
+	RunOn        []string          `meddler:"task_run_on,json"`
 }
 
 // TaskStore defines storage for scheduled Tasks.
@@ -39,13 +41,18 @@ type TaskStore interface {
 // ensures the task Queue can be restored when the system starts.
 func WithTaskStore(q queue.Queue, s TaskStore) queue.Queue {
 	tasks, _ := s.TaskList()
+	toEnqueue := []*queue.Task{}
 	for _, task := range tasks {
-		q.Push(context.Background(), &queue.Task{
+		toEnqueue = append(toEnqueue, &queue.Task{
-			ID:     task.ID,
+			ID:           task.ID,
-			Data:   task.Data,
+			Data:         task.Data,
-			Labels: task.Labels,
+			Labels:       task.Labels,
+			Dependencies: task.Dependencies,
+			RunOn:        task.RunOn,
+			DepStatus:    make(map[string]bool),
 		})
 	}
+	q.PushAtOnce(context.Background(), toEnqueue)
 	return &persistentQueue{q, s}
 }
 
@@ -54,12 +61,14 @@ type persistentQueue struct {
 	store TaskStore
 }
 
-// Push pushes an task to the tail of this queue.
+// Push pushes a task to the tail of this queue.
 func (q *persistentQueue) Push(c context.Context, task *queue.Task) error {
 	q.store.TaskInsert(&Task{
-		ID:     task.ID,
+		ID:           task.ID,
-		Data:   task.Data,
+		Data:         task.Data,
-		Labels: task.Labels,
+		Labels:       task.Labels,
+		Dependencies: task.Dependencies,
+		RunOn:        task.RunOn,
 	})
 	err := q.Queue.Push(c, task)
 	if err != nil {
@@ -68,6 +77,26 @@ func (q *persistentQueue) Push(c context.Context, task *queue.Task) error {
 	return err
 }
 
+// Push pushes multiple tasks to the tail of this queue.
+func (q *persistentQueue) PushAtOnce(c context.Context, tasks []*queue.Task) error {
+	for _, task := range tasks {
+		q.store.TaskInsert(&Task{
+			ID:           task.ID,
+			Data:         task.Data,
+			Labels:       task.Labels,
+			Dependencies: task.Dependencies,
+			RunOn:        task.RunOn,
+		})
+	}
+	err := q.Queue.PushAtOnce(c, tasks)
+	if err != nil {
+		for _, task := range tasks {
+			q.store.TaskDelete(task.ID)
+		}
+	}
+	return err
+}
+
 // Poll retrieves and removes a task head of this queue.
 func (q *persistentQueue) Poll(c context.Context, f queue.Filter) (*queue.Task, error) {
 	task, err := q.Queue.Poll(c, f)

model/repo.go

@@ -55,6 +55,7 @@ type Repo struct {
 	Config      string `json:"config_file"              meddler:"repo_config_path"`
 	Hash        string `json:"-"                        meddler:"repo_hash"`
 	Perm        *Perm  `json:"-"                        meddler:"-"`
+	Fallback    bool   `json:"fallback"                 meddler:"repo_fallback"`
 }
 
 func (r *Repo) ResetVisibility() {
@@ -105,4 +106,5 @@ type RepoPatch struct {
 	AllowDeploy  *bool   `json:"allow_deploy,omitempty"`
 	AllowTag     *bool   `json:"allow_tag,omitempty"`
 	BuildCounter *int    `json:"build_counter,omitempty"`
+	Fallback     *bool   `json:"fallback,omitempty"`
 }

remote/bitbucket/bitbucket.go

@@ -202,20 +202,19 @@ func (c *config) Perm(u *model.User, owner, name string) (*model.Perm, error) {
 
 // File fetches the file from the Bitbucket repository and returns its contents.
 func (c *config) File(u *model.User, r *model.Repo, b *model.Build, f string) ([]byte, error) {
-	return c.FileRef(u, r, b.Commit, f)
+	config, err := c.newClient(u).FindSource(r.Owner, r.Name, b.Commit, f)
-}
-
-// FileRef fetches the file from the Bitbucket repository and returns its contents.
-func (c *config) FileRef(u *model.User, r *model.Repo, ref, f string) ([]byte, error) {
-	config, err := c.newClient(u).FindSource(r.Owner, r.Name, ref, f)
 	if err != nil {
 		return nil, err
 	}
 	return []byte(config.Data), err
 }
 
+func (c *config) Dir(u *model.User, r *model.Repo, b *model.Build, f string) ([]*remote.FileMeta, error) {
+	return nil, fmt.Errorf("Not implemented")
+}
+
 // Status creates a build status for the Bitbucket commit.
-func (c *config) Status(u *model.User, r *model.Repo, b *model.Build, link string) error {
+func (c *config) Status(u *model.User, r *model.Repo, b *model.Build, link string, proc *model.Proc) error {
 	status := internal.BuildStatus{
 		State: convertStatus(b.Status),
 		Desc:  convertDesc(b.Status),

remote/bitbucket/bitbucket_test.go

@@ -283,7 +283,7 @@ func Test_bitbucket(t *testing.T) {
 		})
 
 		g.It("Should update the status", func() {
-			err := c.Status(fakeUser, fakeRepo, fakeBuild, "http://127.0.0.1")
+			err := c.Status(fakeUser, fakeRepo, fakeBuild, "http://127.0.0.1", nil)
 			g.Assert(err == nil).IsTrue()
 		})
 

remote/bitbucketserver/bitbucketserver.go

@@ -179,14 +179,12 @@ func (c *Config) File(u *model.User, r *model.Repo, b *model.Build, f string) ([
 	return client.FindFileForRepo(r.Owner, r.Name, f, b.Ref)
 }
 
-func (c *Config) FileRef(u *model.User, r *model.Repo, ref, f string) ([]byte, error) {
+func (c *Config) Dir(u *model.User, r *model.Repo, b *model.Build, f string) ([]*remote.FileMeta, error) {
-	client := internal.NewClientWithToken(c.URL, c.Consumer, u.Token)
+	return nil, fmt.Errorf("Not implemented")
-
-	return client.FindFileForRepo(r.Owner, r.Name, f, ref)
 }
 
 // Status is not supported by the bitbucketserver driver.
-func (c *Config) Status(u *model.User, r *model.Repo, b *model.Build, link string) error {
+func (c *Config) Status(u *model.User, r *model.Repo, b *model.Build, link string, proc *model.Proc) error {
 	status := internal.BuildStatus{
 		State: convertStatus(b.Status),
 		Desc:  convertDesc(b.Status),

remote/coding/coding.go

@@ -238,18 +238,12 @@ func (c *Coding) File(u *model.User, r *model.Repo, b *model.Build, f string) ([
 	return data, nil
 }
 
-// FileRef fetches a file from the remote repository for the given ref
+func (c *Coding) Dir(u *model.User, r *model.Repo, b *model.Build, f string) ([]*remote.FileMeta, error) {
-// and returns in string format.
+	return nil, fmt.Errorf("Not implemented")
-func (c *Coding) FileRef(u *model.User, r *model.Repo, ref, f string) ([]byte, error) {
-	data, err := c.newClient(u).GetFile(r.Owner, r.Name, ref, f)
-	if err != nil {
-		return nil, err
-	}
-	return data, nil
 }
 
 // Status sends the commit status to the remote system.
-func (c *Coding) Status(u *model.User, r *model.Repo, b *model.Build, link string) error {
+func (c *Coding) Status(u *model.User, r *model.Repo, b *model.Build, link string, proc *model.Proc) error {
 	// EMPTY: not implemented in Coding OAuth API
 	return nil
 }

remote/coding/coding_test.go

@@ -184,11 +184,6 @@ func Test_coding(t *testing.T) {
 				g.Assert(err == nil).IsTrue()
 				g.Assert(string(data)).Equal("pipeline:\n  test:\n    image: golang:1.6\n    commands:\n      - go test\n")
 			})
-			g.It("Should return file for specified ref", func() {
-				data, err := c.FileRef(fakeUser, fakeRepo, "master", ".drone.yml")
-				g.Assert(err == nil).IsTrue()
-				g.Assert(string(data)).Equal("pipeline:\n  test:\n    image: golang:1.6\n    commands:\n      - go test\n")
-			})
 		})
 
 		g.Describe("When requesting a netrc config", func() {

remote/gerrit/gerrit.go

@@ -103,13 +103,12 @@ func (c *client) File(u *model.User, r *model.Repo, b *model.Build, f string) ([
 	return nil, nil
 }
 
-// File is not supported by the Gerrit driver.
+func (c *client) Dir(u *model.User, r *model.Repo, b *model.Build, f string) ([]*remote.FileMeta, error) {
-func (c *client) FileRef(u *model.User, r *model.Repo, ref, f string) ([]byte, error) {
+	return nil, fmt.Errorf("Not implemented")
-	return nil, nil
 }
 
 // Status is not supported by the Gogs driver.
-func (c *client) Status(u *model.User, r *model.Repo, b *model.Build, link string) error {
+func (c *client) Status(u *model.User, r *model.Repo, b *model.Build, link string, proc *model.Proc) error {
 	return nil
 }
 

remote/gitea/gitea.go

@@ -249,13 +249,12 @@ func (c *client) File(u *model.User, r *model.Repo, b *model.Build, f string) ([
 	return cfg, err
 }
 
-// FileRef fetches the file from the Gitea repository and returns its contents.
+func (c *client) Dir(u *model.User, r *model.Repo, b *model.Build, f string) ([]*remote.FileMeta, error) {
-func (c *client) FileRef(u *model.User, r *model.Repo, ref, f string) ([]byte, error) {
+	return nil, fmt.Errorf("Not implemented")
-	return c.newClientToken(u.Token).GetFile(r.Owner, r.Name, ref, f)
 }
 
 // Status is supported by the Gitea driver.
-func (c *client) Status(u *model.User, r *model.Repo, b *model.Build, link string) error {
+func (c *client) Status(u *model.User, r *model.Repo, b *model.Build, link string, proc *model.Proc) error {
 	client := c.newClientToken(u.Token)
 
 	status := getStatus(b.Status)

remote/gitea/gitea_test.go

@@ -18,10 +18,10 @@ import (
 	"net/http/httptest"
 	"testing"
 
-	"github.com/laszlocph/drone-oss-08/model"
-	"github.com/laszlocph/drone-oss-08/remote/gitea/fixtures"
 	"github.com/franela/goblin"
 	"github.com/gin-gonic/gin"
+	"github.com/laszlocph/drone-oss-08/model"
+	"github.com/laszlocph/drone-oss-08/remote/gitea/fixtures"
 )
 
 func Test_gitea(t *testing.T) {
@@ -149,7 +149,7 @@ func Test_gitea(t *testing.T) {
 		})
 
 		g.It("Should return nil from send build status", func() {
-			err := c.Status(fakeUser, fakeRepo, fakeBuild, "http://gitea.io")
+			err := c.Status(fakeUser, fakeRepo, fakeBuild, "http://gitea.io", nil)
 			g.Assert(err == nil).IsTrue()
 		})
 

remote/github/convert.go

@@ -51,7 +51,7 @@ const (
 // GitHub commit status.
 func convertStatus(status string) string {
 	switch status {
-	case model.StatusPending, model.StatusRunning, model.StatusBlocked:
+	case model.StatusPending, model.StatusRunning, model.StatusBlocked, model.StatusSkipped:
 		return statusPending
 	case model.StatusFailure, model.StatusDeclined:
 		return statusFailure

remote/github/github.go

@@ -23,6 +23,7 @@ import (
 	"regexp"
 	"strconv"
 	"strings"
+	"sync"
 
 	"github.com/laszlocph/drone-oss-08/model"
 	"github.com/laszlocph/drone-oss-08/remote"
@@ -225,22 +226,77 @@ func (c *client) Perm(u *model.User, owner, name string) (*model.Perm, error) {
 
 // File fetches the file from the GitHub repository and returns its contents.
 func (c *client) File(u *model.User, r *model.Repo, b *model.Build, f string) ([]byte, error) {
-	return c.FileRef(u, r, b.Commit, f)
-}
-
-// FileRef fetches the file from the GitHub repository and returns its contents.
-func (c *client) FileRef(u *model.User, r *model.Repo, ref, f string) ([]byte, error) {
 	client := c.newClientToken(u.Token)
 
 	opts := new(github.RepositoryContentGetOptions)
-	opts.Ref = ref
+	opts.Ref = b.Commit
 	data, _, _, err := client.Repositories.GetContents(r.Owner, r.Name, f, opts)
 	if err != nil {
 		return nil, err
 	}
+	if data == nil {
+		return nil, fmt.Errorf("%s is a folder not a file use Dir(..)", f)
+	}
 	return data.Decode()
 }
 
+func (c *client) Dir(u *model.User, r *model.Repo, b *model.Build, f string) ([]*remote.FileMeta, error) {
+	client := c.newClientToken(u.Token)
+
+	opts := new(github.RepositoryContentGetOptions)
+	opts.Ref = b.Commit
+	_, data, _, err := client.Repositories.GetContents(r.Owner, r.Name, f, opts)
+	if err != nil {
+		return nil, err
+	}
+
+	fc := make(chan *remote.FileMeta)
+	errc := make(chan error)
+
+	wg := &sync.WaitGroup{}
+	wg.Add(len(data))
+
+	for _, file := range data {
+		go func(path string) {
+			content, err := c.File(u, r, b, path)
+			if err != nil {
+				errc <- err
+			} else {
+				fc <- &remote.FileMeta{
+					Name: path,
+					Data: content,
+				}
+			}
+		}(f + "/" + *file.Name)
+	}
+
+	var files []*remote.FileMeta
+	var errors []error
+
+	go func() {
+		for {
+			select {
+			case err, open := <-errc:
+				if open {
+					errors = append(errors, err)
+					wg.Done()
+				}
+			case fileMeta, open := <-fc:
+				if open {
+					files = append(files, fileMeta)
+					wg.Done()
+				}
+			}
+		}
+	}()
+
+	wg.Wait()
+	close(fc)
+	close(errc)
+
+	return files, nil
+}
+
 // Netrc returns a netrc file capable of authenticating GitHub requests and
 // cloning GitHub repositories. The netrc will use the global machine account
 // when configured.
@@ -374,17 +430,17 @@ func matchingHooks(hooks []github.Hook, rawurl string) *github.Hook {
 
 // Status sends the commit status to the remote system.
 // An example would be the GitHub pull request status.
-func (c *client) Status(u *model.User, r *model.Repo, b *model.Build, link string) error {
+func (c *client) Status(u *model.User, r *model.Repo, b *model.Build, link string, proc *model.Proc) error {
 	client := c.newClientToken(u.Token)
 	switch b.Event {
 	case "deployment":
 		return deploymentStatus(client, r, b, link)
 	default:
-		return repoStatus(client, r, b, link, c.Context)
+		return repoStatus(client, r, b, link, c.Context, proc)
 	}
 }
 
-func repoStatus(client *github.Client, r *model.Repo, b *model.Build, link, ctx string) error {
+func repoStatus(client *github.Client, r *model.Repo, b *model.Build, link, ctx string, proc *model.Proc) error {
 	context := ctx
 	switch b.Event {
 	case model.EventPull:
@@ -395,10 +451,19 @@ func repoStatus(client *github.Client, r *model.Repo, b *model.Build, link, ctx
 		}
 	}
 
+	status := github.String(convertStatus(b.Status))
+	desc := github.String(convertDesc(b.Status))
+
+	if proc != nil {
+		context += "/" + proc.Name
+		status = github.String(convertStatus(proc.State))
+		desc = github.String(convertDesc(proc.State))
+	}
+
 	data := github.RepoStatus{
 		Context:     github.String(context),
-		State:       github.String(convertStatus(b.Status)),
+		State:       status,
-		Description: github.String(convertDesc(b.Status)),
+		Description: desc,
 		TargetURL:   github.String(link),
 	}
 	_, _, err := client.Repositories.CreateStatus(r.Owner, r.Name, b.Commit, &data)

remote/gitlab/gitlab.go

@@ -325,28 +325,27 @@ func (g *Gitlab) Perm(u *model.User, owner, name string) (*model.Perm, error) {
 
 // File fetches a file from the remote repository and returns in string format.
 func (g *Gitlab) File(user *model.User, repo *model.Repo, build *model.Build, f string) ([]byte, error) {
-	return g.FileRef(user, repo, build.Commit, f)
+	var client = NewClient(g.URL, user.Token, g.SkipVerify)
-}
+	id, err := GetProjectId(g, client, repo.Owner, repo.Name)
-
-// FileRef fetches the file from the GitHub repository and returns its contents.
-func (g *Gitlab) FileRef(u *model.User, r *model.Repo, ref, f string) ([]byte, error) {
-	var client = NewClient(g.URL, u.Token, g.SkipVerify)
-	id, err := GetProjectId(g, client, r.Owner, r.Name)
 	if err != nil {
 		return nil, err
 	}
 
-	out, err := client.RepoRawFileRef(id, ref, f)
+	out, err := client.RepoRawFileRef(id, build.Commit, f)
 	if err != nil {
 		return nil, err
 	}
 	return out, err
 }
 
+func (c *Gitlab) Dir(u *model.User, r *model.Repo, b *model.Build, f string) ([]*remote.FileMeta, error) {
+	return nil, fmt.Errorf("Not implemented")
+}
+
 // NOTE Currently gitlab doesn't support status for commits and events,
 //      also if we want get MR status in gitlab we need implement a special plugin for gitlab,
 //      gitlab uses API to fetch build status on client side. But for now we skip this.
-func (g *Gitlab) Status(u *model.User, repo *model.Repo, b *model.Build, link string) error {
+func (g *Gitlab) Status(u *model.User, repo *model.Repo, b *model.Build, link string, proc *model.Proc) error {
 	client := NewClient(g.URL, u.Token, g.SkipVerify)
 
 	status := getStatus(b.Status)

remote/gitlab3/gitlab.go

@@ -338,25 +338,14 @@ func (g *Gitlab) File(user *model.User, repo *model.Repo, build *model.Build, f
 	return out, err
 }
 
-// FileRef fetches the file from the GitHub repository and returns its contents.
+func (c *Gitlab) Dir(u *model.User, r *model.Repo, b *model.Build, f string) ([]*remote.FileMeta, error) {
-func (g *Gitlab) FileRef(u *model.User, r *model.Repo, ref, f string) ([]byte, error) {
+	return nil, fmt.Errorf("Not implemented")
-	var client = NewClient(g.URL, u.Token, g.SkipVerify)
-	id, err := GetProjectId(g, client, r.Owner, r.Name)
-	if err != nil {
-		return nil, err
-	}
-
-	out, err := client.RepoRawFileRef(id, ref, f)
-	if err != nil {
-		return nil, err
-	}
-	return out, err
 }
 
 // NOTE Currently gitlab doesn't support status for commits and events,
 //      also if we want get MR status in gitlab we need implement a special plugin for gitlab,
 //      gitlab uses API to fetch build status on client side. But for now we skip this.
-func (g *Gitlab) Status(u *model.User, repo *model.Repo, b *model.Build, link string) error {
+func (g *Gitlab) Status(u *model.User, repo *model.Repo, b *model.Build, link string, proc *model.Proc) error {
 	client := NewClient(g.URL, u.Token, g.SkipVerify)
 
 	status := getStatus(b.Status)

remote/gogs/gogs.go

@@ -22,9 +22,9 @@ import (
 	"net/url"
 	"strings"
 
+	"github.com/gogits/go-gogs-client"
 	"github.com/laszlocph/drone-oss-08/model"
 	"github.com/laszlocph/drone-oss-08/remote"
-	"github.com/gogits/go-gogs-client"
 )
 
 // Opts defines configuration options.
@@ -202,13 +202,12 @@ func (c *client) File(u *model.User, r *model.Repo, b *model.Build, f string) ([
 	return cfg, err
 }
 
-// FileRef fetches the file from the Gogs repository and returns its contents.
+func (c *client) Dir(u *model.User, r *model.Repo, b *model.Build, f string) ([]*remote.FileMeta, error) {
-func (c *client) FileRef(u *model.User, r *model.Repo, ref, f string) ([]byte, error) {
+	return nil, fmt.Errorf("Not implemented")
-	return c.newClientToken(u.Token).GetFile(r.Owner, r.Name, ref, f)
 }
 
 // Status is not supported by the Gogs driver.
-func (c *client) Status(u *model.User, r *model.Repo, b *model.Build, link string) error {
+func (c *client) Status(u *model.User, r *model.Repo, b *model.Build, link string, proc *model.Proc) error {
 	return nil
 }
 

remote/gogs/gogs_test.go

@@ -163,7 +163,7 @@ func Test_gogs(t *testing.T) {
 
 		g.It("Should return no-op for usupporeted features", func() {
 			_, err1 := c.Auth("octocat", "4vyW6b49Z")
-			err2 := c.Status(nil, nil, nil, "")
+			err2 := c.Status(nil, nil, nil, "", nil)
 			err3 := c.Deactivate(nil, nil, "")
 			g.Assert(err1 != nil).IsTrue()
 			g.Assert(err2 == nil).IsTrue()

remote/mock/remote.go

@@ -1,25 +1,11 @@
-// Copyright 2018 Drone.IO Inc.
+// Code generated by mockery v1.0.0. DO NOT EDIT.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
 
-package mock
+package mocks
 
-import (
+import http "net/http"
-	"net/http"
+import mock "github.com/stretchr/testify/mock"
-
+import model "github.com/laszlocph/drone-oss-08/model"
-	"github.com/laszlocph/drone-oss-08/model"
+import remote "github.com/laszlocph/drone-oss-08/remote"
-	"github.com/stretchr/testify/mock"
-)
 
 // Remote is an autogenerated mock type for the Remote type
 type Remote struct {
@@ -75,6 +61,29 @@ func (_m *Remote) Deactivate(u *model.User, r *model.Repo, link string) error {
 	return r0
 }
 
+// Dir provides a mock function with given fields: u, r, b, f
+func (_m *Remote) Dir(u *model.User, r *model.Repo, b *model.Build, f string) ([]*remote.FileMeta, error) {
+	ret := _m.Called(u, r, b, f)
+
+	var r0 []*remote.FileMeta
+	if rf, ok := ret.Get(0).(func(*model.User, *model.Repo, *model.Build, string) []*remote.FileMeta); ok {
+		r0 = rf(u, r, b, f)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]*remote.FileMeta)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(*model.User, *model.Repo, *model.Build, string) error); ok {
+		r1 = rf(u, r, b, f)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // File provides a mock function with given fields: u, r, b, f
 func (_m *Remote) File(u *model.User, r *model.Repo, b *model.Build, f string) ([]byte, error) {
 	ret := _m.Called(u, r, b, f)
@@ -98,29 +107,6 @@ func (_m *Remote) File(u *model.User, r *model.Repo, b *model.Build, f string) (
 	return r0, r1
 }
 
-// FileRef provides a mock function with given fields: u, r, ref, f
-func (_m *Remote) FileRef(u *model.User, r *model.Repo, ref string, f string) ([]byte, error) {
-	ret := _m.Called(u, r, ref, f)
-
-	var r0 []byte
-	if rf, ok := ret.Get(0).(func(*model.User, *model.Repo, string, string) []byte); ok {
-		r0 = rf(u, r, ref, f)
-	} else {
-		if ret.Get(0) != nil {
-			r0 = ret.Get(0).([]byte)
-		}
-	}
-
-	var r1 error
-	if rf, ok := ret.Get(1).(func(*model.User, *model.Repo, string, string) error); ok {
-		r1 = rf(u, r, ref, f)
-	} else {
-		r1 = ret.Error(1)
-	}
-
-	return r0, r1
-}
-
 // Hook provides a mock function with given fields: r
 func (_m *Remote) Hook(r *http.Request) (*model.Repo, *model.Build, error) {
 	ret := _m.Called(r)
@@ -246,15 +232,15 @@ func (_m *Remote) Repo(u *model.User, owner string, repo string) (*model.Repo, e
 }
 
 // Repos provides a mock function with given fields: u
-func (_m *Remote) Repos(u *model.User) ([]*model.RepoLite, error) {
+func (_m *Remote) Repos(u *model.User) ([]*model.Repo, error) {
 	ret := _m.Called(u)
 
-	var r0 []*model.RepoLite
+	var r0 []*model.Repo
-	if rf, ok := ret.Get(0).(func(*model.User) []*model.RepoLite); ok {
+	if rf, ok := ret.Get(0).(func(*model.User) []*model.Repo); ok {
 		r0 = rf(u)
 	} else {
 		if ret.Get(0) != nil {
-			r0 = ret.Get(0).([]*model.RepoLite)
+			r0 = ret.Get(0).([]*model.Repo)
 		}
 	}
 
@@ -282,29 +268,6 @@ func (_m *Remote) Status(u *model.User, r *model.Repo, b *model.Build, link stri
 	return r0
 }
 
-// TeamPerm provides a mock function with given fields: u, org
-func (_m *Remote) TeamPerm(u *model.User, org string) (*model.Perm, error) {
-	ret := _m.Called(u, org)
-
-	var r0 *model.Perm
-	if rf, ok := ret.Get(0).(func(*model.User, string) *model.Perm); ok {
-		r0 = rf(u, org)
-	} else {
-		if ret.Get(0) != nil {
-			r0 = ret.Get(0).(*model.Perm)
-		}
-	}
-
-	var r1 error
-	if rf, ok := ret.Get(1).(func(*model.User, string) error); ok {
-		r1 = rf(u, org)
-	} else {
-		r1 = ret.Error(1)
-	}
-
-	return r0, r1
-}
-
 // Teams provides a mock function with given fields: u
 func (_m *Remote) Teams(u *model.User) ([]*model.Team, error) {
 	ret := _m.Called(u)

remote/remote.go

@@ -18,7 +18,6 @@ package remote
 
 import (
 	"net/http"
-	"time"
 
 	"github.com/laszlocph/drone-oss-08/model"
 
@@ -51,13 +50,12 @@ type Remote interface {
 	// format.
 	File(u *model.User, r *model.Repo, b *model.Build, f string) ([]byte, error)
 
-	// FileRef fetches a file from the remote repository for the given ref
+	// Dir fetches a folder from the remote repository
-	// and returns in string format.
+	Dir(u *model.User, r *model.Repo, b *model.Build, f string) ([]*FileMeta, error)
-	FileRef(u *model.User, r *model.Repo, ref, f string) ([]byte, error)
 
 	// Status sends the commit status to the remote system.
 	// An example would be the GitHub pull request status.
-	Status(u *model.User, r *model.Repo, b *model.Build, link string) error
+	Status(u *model.User, r *model.Repo, b *model.Build, link string, proc *model.Proc) error
 
 	// Netrc returns a .netrc file that can be used to clone
 	// private repositories from a remote system.
@@ -75,6 +73,18 @@ type Remote interface {
 	Hook(r *http.Request) (*model.Repo, *model.Build, error)
 }
 
+// FileMeta represents a file in version control
+type FileMeta struct {
+	Name string
+	Data []byte
+}
+
+type ByName []*FileMeta
+
+func (a ByName) Len() int           { return len(a) }
+func (a ByName) Less(i, j int) bool { return a[i].Name < a[j].Name }
+func (a ByName) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
+
 // Refresher refreshes an oauth token and expiration for the given user. It
 // returns true if the token was refreshed, false if the token was not refreshed,
 // and error if it failed to refersh.
@@ -115,22 +125,10 @@ func Perm(c context.Context, u *model.User, owner, repo string) (*model.Perm, er
 	return FromContext(c).Perm(u, owner, repo)
 }
 
-// File fetches a file from the remote repository and returns in string format.
-func File(c context.Context, u *model.User, r *model.Repo, b *model.Build, f string) (out []byte, err error) {
-	for i := 0; i < 12; i++ {
-		out, err = FromContext(c).File(u, r, b, f)
-		if err == nil {
-			return
-		}
-		time.Sleep(5 * time.Second)
-	}
-	return
-}
-
 // Status sends the commit status to the remote system.
 // An example would be the GitHub pull request status.
-func Status(c context.Context, u *model.User, r *model.Repo, b *model.Build, link string) error {
+func Status(c context.Context, u *model.User, r *model.Repo, b *model.Build, link string, proc *model.Proc) error {
-	return FromContext(c).Status(u, r, b, link)
+	return FromContext(c).Status(u, r, b, link, proc)
 }
 
 // Netrc returns a .netrc file that can be used to clone
@@ -168,18 +166,3 @@ func Refresh(c context.Context, u *model.User) (bool, error) {
 	}
 	return refresher.Refresh(u)
 }
-
-// FileBackoff fetches the file using an exponential backoff.
-// TODO replace this with a proper backoff
-func FileBackoff(remote Remote, u *model.User, r *model.Repo, b *model.Build, f string) (out []byte, err error) {
-	for i := 0; i < 5; i++ {
-		select {
-		case <-time.After(time.Second * time.Duration(i)):
-			out, err = remote.File(u, r, b, f)
-			if err == nil {
-				return
-			}
-		}
-	}
-	return
-}

server/build.go

@@ -156,13 +156,12 @@ func GetProcLogs(c *gin.Context) {
 	io.Copy(c.Writer, rc)
 }
 
+// DeleteBuild cancels a build
 func DeleteBuild(c *gin.Context) {
 	repo := session.Repo(c)
 
-	// parse the build number and job sequence number from
+	// parse the build number from the request parameter.
-	// the repquest parameter.
 	num, _ := strconv.Atoi(c.Params.ByName("number"))
-	seq, _ := strconv.Atoi(c.Params.ByName("job"))
 
 	build, err := store.GetBuildNumber(c, repo, num)
 	if err != nil {
@@ -170,27 +169,40 @@ func DeleteBuild(c *gin.Context) {
 		return
 	}
 
-	proc, err := store.FromContext(c).ProcFind(build, seq)
+	procs, err := store.FromContext(c).ProcList(build)
 	if err != nil {
 		c.AbortWithError(404, err)
 		return
 	}
 
-	if proc.State != model.StatusRunning {
+	cancelled := false
+	for _, proc := range procs {
+		if proc.PPID != 0 {
+			continue
+		}
+
+		if proc.State != model.StatusRunning && proc.State != model.StatusPending {
+			continue
+		}
+
+		proc.State = model.StatusKilled
+		proc.Stopped = time.Now().Unix()
+		if proc.Started == 0 {
+			proc.Started = proc.Stopped
+		}
+		proc.ExitCode = 137
+		// TODO cancel child procs
+		store.FromContext(c).ProcUpdate(proc)
+
+		Config.Services.Queue.Error(context.Background(), fmt.Sprint(proc.ID), queue.ErrCancel)
+		cancelled = true
+	}
+
+	if !cancelled {
 		c.String(400, "Cannot cancel a non-running build")
 		return
 	}
 
-	proc.State = model.StatusKilled
-	proc.Stopped = time.Now().Unix()
-	if proc.Started == 0 {
-		proc.Started = proc.Stopped
-	}
-	proc.ExitCode = 137
-	// TODO cancel child procs
-	store.FromContext(c).ProcUpdate(proc)
-
-	Config.Services.Queue.Error(context.Background(), fmt.Sprint(proc.ID), queue.ErrCancel)
 	c.String(204, "")
 }
 
@@ -268,7 +280,7 @@ func PostApproval(c *gin.Context) {
 	build.Reviewer = user.Login
 
 	// fetch the build file from the database
-	conf, err := Config.Storage.Config.ConfigLoad(build.ConfigID)
+	configs, err := Config.Storage.Config.ConfigsForBuild(build.ID)
 	if err != nil {
 		logrus.Errorf("failure to get build config for %s. %s", repo.FullName, err)
 		c.AbortWithError(404, err)
@@ -307,13 +319,10 @@ func PostApproval(c *gin.Context) {
 		}
 	}
 
-	defer func() {
+	var yamls []*remote.FileMeta
-		uri := fmt.Sprintf("%s/%s/%d", httputil.GetURL(c.Request), repo.FullName, build.Number)
+	for _, y := range configs {
-		err = remote_.Status(user, repo, build, uri)
+		yamls = append(yamls, &remote.FileMeta{Data: []byte(y.Data), Name: y.Name})
-		if err != nil {
+	}
-			logrus.Errorf("error setting commit status for %s/%d: %v", repo.FullName, build.Number, err)
-		}
-	}()
 
 	b := procBuilder{
 		Repo:  repo,
@@ -323,7 +332,7 @@ func PostApproval(c *gin.Context) {
 		Secs:  secs,
 		Regs:  regs,
 		Link:  httputil.GetURL(c.Request),
-		Yaml:  conf.Data,
+		Yamls: yamls,
 		Envs:  envs,
 	}
 	buildItems, err := b.Build()
@@ -336,12 +345,25 @@ func PostApproval(c *gin.Context) {
 		return
 	}
 
-	setBuildProcs(build, buildItems)
 	err = store.FromContext(c).ProcCreate(build.Procs)
 	if err != nil {
 		logrus.Errorf("error persisting procs %s/%d: %s", repo.FullName, build.Number, err)
 	}
 
+	defer func() {
+		for _, item := range buildItems {
+			uri := fmt.Sprintf("%s/%s/%d", httputil.GetURL(c.Request), repo.FullName, build.Number)
+			if len(buildItems) > 1 {
+				err = remote_.Status(user, repo, build, uri, item.Proc)
+			} else {
+				err = remote_.Status(user, repo, build, uri, nil)
+			}
+			if err != nil {
+				logrus.Errorf("error setting commit status for %s/%d: %v", repo.FullName, build.Number, err)
+			}
+		}
+	}()
+
 	publishToTopic(c, build, repo)
 	queueBuild(build, repo, buildItems)
 }
@@ -376,7 +398,7 @@ func PostDecline(c *gin.Context) {
 	}
 
 	uri := fmt.Sprintf("%s/%s/%d", httputil.GetURL(c.Request), repo.FullName, build.Number)
-	err = remote_.Status(user, repo, build, uri)
+	err = remote_.Status(user, repo, build, uri, nil)
 	if err != nil {
 		logrus.Errorf("error setting commit status for %s/%d: %v", repo.FullName, build.Number, err)
 	}
@@ -436,7 +458,7 @@ func PostBuild(c *gin.Context) {
 	}
 
 	// fetch the .drone.yml file from the database
-	conf, err := Config.Storage.Config.ConfigLoad(build.ConfigID)
+	configs, err := Config.Storage.Config.ConfigsForBuild(build.ID)
 	if err != nil {
 		logrus.Errorf("failure to get build config for %s. %s", repo.FullName, err)
 		c.AbortWithError(404, err)
@@ -474,6 +496,13 @@ func PostBuild(c *gin.Context) {
 		return
 	}
 
+	err = persistBuildConfigs(configs, build.ID)
+	if err != nil {
+		logrus.Errorf("failure to persist build config for %s. %s", repo.FullName, err)
+		c.AbortWithError(500, err)
+		return
+	}
+
 	// Read query string parameters into buildParams, exclude reserved params
 	var buildParams = map[string]string{}
 	for key, val := range c.Request.URL.Query() {
@@ -504,6 +533,11 @@ func PostBuild(c *gin.Context) {
 		}
 	}
 
+	var yamls []*remote.FileMeta
+	for _, y := range configs {
+		yamls = append(yamls, &remote.FileMeta{Data: []byte(y.Data), Name: y.Name})
+	}
+
 	b := procBuilder{
 		Repo:  repo,
 		Curr:  build,
@@ -512,7 +546,7 @@ func PostBuild(c *gin.Context) {
 		Secs:  secs,
 		Regs:  regs,
 		Link:  httputil.GetURL(c.Request),
-		Yaml:  conf.Data,
+		Yamls: yamls,
 		Envs:  buildParams,
 	}
 	buildItems, err := b.Build()
@@ -525,8 +559,6 @@ func PostBuild(c *gin.Context) {
 		return
 	}
 
-	setBuildProcs(build, buildItems)
-
 	err = store.FromContext(c).ProcCreate(build.Procs)
 	if err != nil {
 		logrus.Errorf("cannot restart %s#%d: %s", repo.FullName, build.Number, err)
@@ -582,6 +614,20 @@ func DeleteBuildLogs(c *gin.Context) {
 	c.String(204, "")
 }
 
+func persistBuildConfigs(configs []*model.Config, buildID int64) error {
+	for _, conf := range configs {
+		buildConfig := &model.BuildConfig{
+			ConfigID: conf.ID,
+			BuildID:  buildID,
+		}
+		err := Config.Storage.Config.BuildConfigCreate(buildConfig)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 var deleteStr = `[
 	{
 	  "proc": %q,

server/configFetcher.go

@@ -0,0 +1,53 @@
+package server
+
+import (
+	"strings"
+	"time"
+
+	"github.com/laszlocph/drone-oss-08/model"
+	"github.com/laszlocph/drone-oss-08/remote"
+)
+
+type configFetcher struct {
+	remote_ remote.Remote
+	user    *model.User
+	repo    *model.Repo
+	build   *model.Build
+}
+
+func (cf *configFetcher) Fetch() ([]*remote.FileMeta, error) {
+	for i := 0; i < 5; i++ {
+		select {
+		case <-time.After(time.Second * time.Duration(i)):
+			// either a file
+			file, fileerr := cf.remote_.File(cf.user, cf.repo, cf.build, cf.repo.Config)
+			if fileerr == nil {
+				return []*remote.FileMeta{&remote.FileMeta{
+					Name: cf.repo.Config,
+					Data: file,
+				}}, nil
+			}
+
+			// or a folder
+			dir, direrr := cf.remote_.Dir(cf.user, cf.repo, cf.build, strings.TrimSuffix(cf.repo.Config, "/"))
+
+			if direrr == nil {
+				return dir, nil
+			} else if !cf.repo.Fallback {
+				return nil, direrr
+			}
+
+			// or fallback
+			file, fileerr = cf.remote_.File(cf.user, cf.repo, cf.build, ".drone.yml")
+			if fileerr != nil {
+				return nil, fileerr
+			}
+
+			return []*remote.FileMeta{&remote.FileMeta{
+				Name: cf.repo.Config,
+				Data: file,
+			}}, nil
+		}
+	}
+	return []*remote.FileMeta{}, nil
+}

server/configFetcher_test.go

@@ -0,0 +1,22 @@
+package server
+
+import (
+	"testing"
+
+	"github.com/laszlocph/drone-oss-08/model"
+	"github.com/laszlocph/drone-oss-08/remote/github"
+)
+
+func TestFetchGithub(t *testing.T) {
+	github, err := github.New(github.Opts{URL: "https://github.com"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	configFetcher := &configFetcher{
+		remote_: github,
+		user:    &model.User{Token: "xxx"},
+		repo:    &model.Repo{Owner: "laszlocph", Name: "drone-multipipeline", Config: ".drone"},
+		build:   &model.Build{Commit: "89ab7b2d6bfb347144ac7c557e638ab402848fee"},
+	}
+	configFetcher.Fetch()
+}

server/hook.go

@@ -143,34 +143,21 @@ func PostHook(c *gin.Context) {
 	}
 
 	// fetch the build file from the remote
-	remoteYamlConfig, err := remote.FileBackoff(remote_, user, repo, build, repo.Config)
+	configFetcher := &configFetcher{remote_: remote_, user: user, repo: repo, build: build}
+	remoteYamlConfigs, err := configFetcher.Fetch()
 	if err != nil {
 		logrus.Errorf("error: %s: cannot find %s in %s: %s", repo.FullName, repo.Config, build.Ref, err)
 		c.AbortWithError(404, err)
 		return
 	}
-	conf, err := findOrPersistPipelineConfig(repo, remoteYamlConfig)
+
-	if err != nil {
+	if branchFiltered(build, remoteYamlConfigs) {
-		logrus.Errorf("failure to find or persist build config for %s. %s", repo.FullName, err)
+		c.String(200, "Branch does not match restrictions defined in yaml")
-		c.AbortWithError(500, err)
 		return
 	}
-	build.ConfigID = conf.ID
 
-	// verify that pipeline can be built at all
+	if repo.IsGated { // This feature is not clear to me. Reenabling once better understood
-	parsedPipelineConfig, err := yaml.ParseString(conf.Data)
+		build.Status = model.StatusBlocked
-	if err == nil {
-		if !parsedPipelineConfig.Branches.Match(build.Branch) && build.Event != model.EventTag && build.Event != model.EventDeploy {
-			c.String(200, "Branch does not match restrictions defined in yaml")
-			return
-		}
-	}
-
-	if repo.IsGated {
-		allowed, _ := Config.Services.Senders.SenderAllowed(user, repo, build, conf)
-		if !allowed {
-			build.Status = model.StatusBlocked
-		}
 	}
 
 	// update some build fields
@@ -185,6 +172,16 @@ func PostHook(c *gin.Context) {
 		return
 	}
 
+	// persist the build config for historical correctness, restarts, etc
+	for _, remoteYamlConfig := range remoteYamlConfigs {
+		_, err := findOrPersistPipelineConfig(build, remoteYamlConfig)
+		if err != nil {
+			logrus.Errorf("failure to find or persist build config for %s. %s", repo.FullName, err)
+			c.AbortWithError(500, err)
+			return
+		}
+	}
+
 	c.JSON(200, build)
 
 	if build.Status == model.StatusBlocked {
@@ -218,14 +215,6 @@ func PostHook(c *gin.Context) {
 	// get the previous build so that we can send status change notifications
 	last, _ := store.GetBuildLastBefore(c, repo, build.Branch, build.ID)
 
-	defer func() {
-		uri := fmt.Sprintf("%s/%s/%d", httputil.GetURL(c.Request), repo.FullName, build.Number)
-		err = remote_.Status(user, repo, build, uri)
-		if err != nil {
-			logrus.Errorf("error setting commit status for %s/%d: %v", repo.FullName, build.Number, err)
-		}
-	}()
-
 	b := procBuilder{
 		Repo:  repo,
 		Curr:  build,
@@ -235,7 +224,7 @@ func PostHook(c *gin.Context) {
 		Regs:  regs,
 		Envs:  envs,
 		Link:  httputil.GetURL(c.Request),
-		Yaml:  conf.Data,
+		Yamls: remoteYamlConfigs,
 	}
 	buildItems, err := b.Build()
 	if err != nil {
@@ -247,66 +236,75 @@ func PostHook(c *gin.Context) {
 		return
 	}
 
-	setBuildProcs(build, buildItems)
-
 	err = store.FromContext(c).ProcCreate(build.Procs)
 	if err != nil {
 		logrus.Errorf("error persisting procs %s/%d: %s", repo.FullName, build.Number, err)
 	}
 
+	defer func() {
+		for _, item := range buildItems {
+			uri := fmt.Sprintf("%s/%s/%d", httputil.GetURL(c.Request), repo.FullName, build.Number)
+			if len(buildItems) > 1 {
+				err = remote_.Status(user, repo, build, uri, item.Proc)
+			} else {
+				err = remote_.Status(user, repo, build, uri, nil)
+			}
+			if err != nil {
+				logrus.Errorf("error setting commit status for %s/%d: %v", repo.FullName, build.Number, err)
+			}
+		}
+	}()
+
 	publishToTopic(c, build, repo)
 	queueBuild(build, repo, buildItems)
 }
 
-func findOrPersistPipelineConfig(repo *model.Repo, remoteYamlConfig []byte) (*model.Config, error) {
+func branchFiltered(build *model.Build, remoteYamlConfigs []*remote.FileMeta) bool {
-	sha := shasum(remoteYamlConfig)
+	for _, remoteYamlConfig := range remoteYamlConfigs {
-	conf, err := Config.Storage.Config.ConfigFind(repo, sha)
+		parsedPipelineConfig, err := yaml.ParseString(string(remoteYamlConfig.Data))
+		if err == nil {
+			if !parsedPipelineConfig.Branches.Match(build.Branch) && build.Event != model.EventTag && build.Event != model.EventDeploy {
+			} else {
+				return false
+			}
+		}
+	}
+	return true
+}
+
+func findOrPersistPipelineConfig(build *model.Build, remoteYamlConfig *remote.FileMeta) (*model.Config, error) {
+	sha := shasum(remoteYamlConfig.Data)
+	conf, err := Config.Storage.Config.ConfigFindIdentical(build.RepoID, sha)
 	if err != nil {
 		conf = &model.Config{
-			RepoID: repo.ID,
+			RepoID: build.RepoID,
-			Data:   string(remoteYamlConfig),
+			Data:   string(remoteYamlConfig.Data),
 			Hash:   sha,
+			Name:   sanitizePath(remoteYamlConfig.Name),
 		}
 		err = Config.Storage.Config.ConfigCreate(conf)
 		if err != nil {
 			// retry in case we receive two hooks at the same time
-			conf, err = Config.Storage.Config.ConfigFind(repo, sha)
+			conf, err = Config.Storage.Config.ConfigFindIdentical(build.RepoID, sha)
 			if err != nil {
 				return nil, err
 			}
 		}
 	}
 
+	buildConfig := &model.BuildConfig{
+		ConfigID: conf.ID,
+		BuildID:  build.ID,
+	}
+	err = Config.Storage.Config.BuildConfigCreate(buildConfig)
+	if err != nil {
+		return nil, err
+	}
+
 	return conf, nil
 }
 
-func setBuildProcs(build *model.Build, buildItems []*buildItem) {
+// publishes message to UI clients
-	pcounter := len(buildItems)
-	for _, item := range buildItems {
-		build.Procs = append(build.Procs, item.Proc)
-		item.Proc.BuildID = build.ID
-
-		for _, stage := range item.Config.Stages {
-			var gid int
-			for _, step := range stage.Steps {
-				pcounter++
-				if gid == 0 {
-					gid = pcounter
-				}
-				proc := &model.Proc{
-					BuildID: build.ID,
-					Name:    step.Alias,
-					PID:     pcounter,
-					PPID:    item.Proc.PID,
-					PGID:    gid,
-					State:   model.StatusPending,
-				}
-				build.Procs = append(build.Procs, proc)
-			}
-		}
-	}
-}
-
 func publishToTopic(c *gin.Context, build *model.Build, repo *model.Repo) {
 	message := pubsub.Message{
 		Labels: map[string]string{
@@ -325,7 +323,11 @@ func publishToTopic(c *gin.Context, build *model.Build, repo *model.Repo) {
 }
 
 func queueBuild(build *model.Build, repo *model.Repo, buildItems []*buildItem) {
+	var tasks []*queue.Task
 	for _, item := range buildItems {
+		if item.Proc.State == model.StatusSkipped {
+			continue
+		}
 		task := new(queue.Task)
 		task.ID = fmt.Sprint(item.Proc.ID)
 		task.Labels = map[string]string{}
@@ -334,6 +336,9 @@ func queueBuild(build *model.Build, repo *model.Repo, buildItems []*buildItem) {
 		}
 		task.Labels["platform"] = item.Platform
 		task.Labels["repo"] = repo.FullName
+		task.Dependencies = taskIds(item.DependsOn, buildItems)
+		task.RunOn = item.RunsOn
+		task.DepStatus = make(map[string]bool)
 
 		task.Data, _ = json.Marshal(rpc.Pipeline{
 			ID:      fmt.Sprint(item.Proc.ID),
@@ -342,8 +347,21 @@ func queueBuild(build *model.Build, repo *model.Repo, buildItems []*buildItem) {
 		})
 
 		Config.Services.Logs.Open(context.Background(), task.ID)
-		Config.Services.Queue.Push(context.Background(), task)
+		tasks = append(tasks, task)
 	}
+	Config.Services.Queue.PushAtOnce(context.Background(), tasks)
+}
+
+func taskIds(dependsOn []string, buildItems []*buildItem) []string {
+	taskIds := []string{}
+	for _, dep := range dependsOn {
+		for _, buildItem := range buildItems {
+			if buildItem.Proc.Name == dep {
+				taskIds = append(taskIds, fmt.Sprint(buildItem.Proc.ID))
+			}
+		}
+	}
+	return taskIds
 }
 
 func shasum(raw []byte) string {

server/hook_test.go

@@ -1,45 +0,0 @@
-// Copyright 2018 Drone.IO Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package server
-
-import (
-	"testing"
-
-	"github.com/laszlocph/drone-oss-08/model"
-)
-
-func TestMultilineEnvsubst(t *testing.T) {
-	b := procBuilder{
-		Repo: &model.Repo{},
-		Curr: &model.Build{
-			Message: `aaa
-bbb`,
-		},
-		Last:  &model.Build{},
-		Netrc: &model.Netrc{},
-		Secs:  []*model.Secret{},
-		Regs:  []*model.Registry{},
-		Link:  "",
-		Yaml: `pipeline:
-  xxx:
-    image: scratch
-    yyy: ${DRONE_COMMIT_MESSAGE}
-`,
-	}
-
-	if _, err := b.Build(); err != nil {
-		t.Fatal(err)
-	}
-}

server/procBuilder.go

@@ -18,6 +18,7 @@ import (
 	"fmt"
 	"math/rand"
 	"net/url"
+	"sort"
 	"strings"
 
 	"github.com/drone/envsubst"
@@ -28,6 +29,7 @@ import (
 	"github.com/laszlocph/drone-oss-08/cncd/pipeline/pipeline/frontend/yaml/linter"
 	"github.com/laszlocph/drone-oss-08/cncd/pipeline/pipeline/frontend/yaml/matrix"
 	"github.com/laszlocph/drone-oss-08/model"
+	"github.com/laszlocph/drone-oss-08/remote"
 )
 
 // Takes the hook data and the yaml and returns in internal data model
@@ -39,156 +41,198 @@ type procBuilder struct {
 	Secs  []*model.Secret
 	Regs  []*model.Registry
 	Link  string
-	Yaml  string
+	Yamls []*remote.FileMeta
 	Envs  map[string]string
 }
 
 type buildItem struct {
-	Proc     *model.Proc
+	Proc      *model.Proc
-	Platform string
+	Platform  string
-	Labels   map[string]string
+	Labels    map[string]string
-	Config   *backend.Config
+	DependsOn []string
+	RunsOn    []string
+	Config    *backend.Config
 }
 
 func (b *procBuilder) Build() ([]*buildItem, error) {
-
-	axes, err := matrix.ParseString(b.Yaml)
-	if err != nil {
-		return nil, err
-	}
-	if len(axes) == 0 {
-		axes = append(axes, matrix.Axis{})
-	}
-
 	var items []*buildItem
-	for i, axis := range axes {
-		proc := &model.Proc{
-			BuildID: b.Curr.ID,
-			PID:     i + 1,
-			PGID:    i + 1,
-			State:   model.StatusPending,
-			Environ: axis,
-		}
 
-		metadata := metadataFromStruct(b.Repo, b.Curr, b.Last, proc, b.Link)
+	sort.Sort(remote.ByName(b.Yamls))
-		environ := metadata.Environ()
-		for k, v := range metadata.EnvironDrone() {
-			environ[k] = v
-		}
-		for k, v := range axis {
-			environ[k] = v
-		}
 
-		var secrets []compiler.Secret
+	for j, y := range b.Yamls {
-		for _, sec := range b.Secs {
+		// matrix axes
-			if !sec.Match(b.Curr.Event) {
+		axes, err := matrix.ParseString(string(y.Data))
-				continue
-			}
-			secrets = append(secrets, compiler.Secret{
-				Name:  sec.Name,
-				Value: sec.Value,
-				Match: sec.Images,
-			})
-		}
-
-		y := b.Yaml
-		s, err := envsubst.Eval(y, func(name string) string {
-			env := environ[name]
-			if strings.Contains(env, "\n") {
-				env = fmt.Sprintf("%q", env)
-			}
-			return env
-		})
 		if err != nil {
 			return nil, err
 		}
-		y = s
+		if len(axes) == 0 {
-
+			axes = append(axes, matrix.Axis{})
-		parsed, err := yaml.ParseString(y)
-		if err != nil {
-			return nil, err
-		}
-		metadata.Sys.Arch = parsed.Platform
-		if metadata.Sys.Arch == "" {
-			metadata.Sys.Arch = "linux/amd64"
 		}
 
-		lerr := linter.New(
+		for i, axis := range axes {
-			linter.WithTrusted(b.Repo.IsTrusted),
+			proc := &model.Proc{
-		).Lint(parsed)
+				BuildID: b.Curr.ID,
-		if lerr != nil {
+				PID:     j + i + 1,
-			return nil, lerr
+				PGID:    j + i + 1,
-		}
+				State:   model.StatusPending,
+				Environ: axis,
+				Name:    sanitizePath(y.Name),
+			}
+			b.Curr.Procs = append(b.Curr.Procs, proc)
 
-		var registries []compiler.Registry
+			metadata := metadataFromStruct(b.Repo, b.Curr, b.Last, proc, b.Link)
-		for _, reg := range b.Regs {
+			environ := b.environmentVariables(metadata, axis)
-			registries = append(registries, compiler.Registry{
-				Hostname: reg.Address,
-				Username: reg.Username,
-				Password: reg.Password,
-				Email:    reg.Email,
-			})
-		}
 
-		ir := compiler.New(
+			// substitute vars
-			compiler.WithEnviron(environ),
+			substituted, err := b.envsubst_(string(y.Data), environ)
-			compiler.WithEnviron(b.Envs),
+			if err != nil {
-			compiler.WithEscalated(Config.Pipeline.Privileged...),
+				return nil, err
-			compiler.WithResourceLimit(Config.Pipeline.Limits.MemSwapLimit, Config.Pipeline.Limits.MemLimit, Config.Pipeline.Limits.ShmSize, Config.Pipeline.Limits.CPUQuota, Config.Pipeline.Limits.CPUShares, Config.Pipeline.Limits.CPUSet),
+			}
-			compiler.WithVolumes(Config.Pipeline.Volumes...),
-			compiler.WithNetworks(Config.Pipeline.Networks...),
-			compiler.WithLocal(false),
-			compiler.WithOption(
-				compiler.WithNetrc(
-					b.Netrc.Login,
-					b.Netrc.Password,
-					b.Netrc.Machine,
-				),
-				b.Repo.IsPrivate,
-			),
-			compiler.WithRegistry(registries...),
-			compiler.WithSecret(secrets...),
-			compiler.WithPrefix(
-				fmt.Sprintf(
-					"%d_%d",
-					proc.ID,
-					rand.Int(),
-				),
-			),
-			compiler.WithEnviron(proc.Environ),
-			compiler.WithProxy(),
-			compiler.WithWorkspaceFromURL("/drone", b.Repo.Link),
-			compiler.WithMetadata(metadata),
-		).Compile(parsed)
 
-		// for _, sec := range b.Secs {
+			// parse yaml pipeline
-		// 	if !sec.MatchEvent(b.Curr.Event) {
+			parsed, err := yaml.ParseString(substituted)
-		// 		continue
+			if err != nil {
-		// 	}
+				return nil, err
-		// 	if b.Curr.Verified || sec.SkipVerify {
+			}
-		// 		ir.Secrets = append(ir.Secrets, &backend.Secret{
-		// 			Mask:  sec.Conceal,
-		// 			Name:  sec.Name,
-		// 			Value: sec.Value,
-		// 		})
-		// 	}
-		// }
 
-		item := &buildItem{
+			// lint pipeline
-			Proc:     proc,
+			lerr := linter.New(
-			Config:   ir,
+				linter.WithTrusted(b.Repo.IsTrusted),
-			Labels:   parsed.Labels,
+			).Lint(parsed)
-			Platform: metadata.Sys.Arch,
+			if lerr != nil {
+				return nil, lerr
+			}
+
+			if !parsed.Branches.Match(b.Curr.Branch) {
+				proc.State = model.StatusSkipped
+			}
+
+			metadata.SetPlatform(parsed.Platform)
+
+			ir := b.toInternalRepresentation(parsed, environ, metadata, proc.ID)
+
+			item := &buildItem{
+				Proc:      proc,
+				Config:    ir,
+				Labels:    parsed.Labels,
+				DependsOn: parsed.DependsOn,
+				RunsOn:    parsed.RunsOn,
+				Platform:  metadata.Sys.Arch,
+			}
+			if item.Labels == nil {
+				item.Labels = map[string]string{}
+			}
+			items = append(items, item)
 		}
-		if item.Labels == nil {
-			item.Labels = map[string]string{}
-		}
-		items = append(items, item)
 	}
 
+	setBuildSteps(b.Curr, items)
+
 	return items, nil
 }
 
+func (b *procBuilder) envsubst_(y string, environ map[string]string) (string, error) {
+	return envsubst.Eval(y, func(name string) string {
+		env := environ[name]
+		if strings.Contains(env, "\n") {
+			env = fmt.Sprintf("%q", env)
+		}
+		return env
+	})
+}
+
+func (b *procBuilder) environmentVariables(metadata frontend.Metadata, axis matrix.Axis) map[string]string {
+	environ := metadata.Environ()
+	for k, v := range metadata.EnvironDrone() {
+		environ[k] = v
+	}
+	for k, v := range axis {
+		environ[k] = v
+	}
+	return environ
+}
+
+func (b *procBuilder) toInternalRepresentation(parsed *yaml.Config, environ map[string]string, metadata frontend.Metadata, procID int64) *backend.Config {
+	var secrets []compiler.Secret
+	for _, sec := range b.Secs {
+		if !sec.Match(b.Curr.Event) {
+			continue
+		}
+		secrets = append(secrets, compiler.Secret{
+			Name:  sec.Name,
+			Value: sec.Value,
+			Match: sec.Images,
+		})
+	}
+
+	var registries []compiler.Registry
+	for _, reg := range b.Regs {
+		registries = append(registries, compiler.Registry{
+			Hostname: reg.Address,
+			Username: reg.Username,
+			Password: reg.Password,
+			Email:    reg.Email,
+		})
+	}
+
+	return compiler.New(
+		compiler.WithEnviron(environ),
+		compiler.WithEnviron(b.Envs),
+		compiler.WithEscalated(Config.Pipeline.Privileged...),
+		compiler.WithResourceLimit(Config.Pipeline.Limits.MemSwapLimit, Config.Pipeline.Limits.MemLimit, Config.Pipeline.Limits.ShmSize, Config.Pipeline.Limits.CPUQuota, Config.Pipeline.Limits.CPUShares, Config.Pipeline.Limits.CPUSet),
+		compiler.WithVolumes(Config.Pipeline.Volumes...),
+		compiler.WithNetworks(Config.Pipeline.Networks...),
+		compiler.WithLocal(false),
+		compiler.WithOption(
+			compiler.WithNetrc(
+				b.Netrc.Login,
+				b.Netrc.Password,
+				b.Netrc.Machine,
+			),
+			b.Repo.IsPrivate,
+		),
+		compiler.WithRegistry(registries...),
+		compiler.WithSecret(secrets...),
+		compiler.WithPrefix(
+			fmt.Sprintf(
+				"%d_%d",
+				procID,
+				rand.Int(),
+			),
+		),
+		compiler.WithProxy(),
+		compiler.WithWorkspaceFromURL("/drone", b.Repo.Link),
+		compiler.WithMetadata(metadata),
+	).Compile(parsed)
+}
+
+func setBuildSteps(build *model.Build, buildItems []*buildItem) {
+	pcounter := len(buildItems)
+	for _, item := range buildItems {
+		for _, stage := range item.Config.Stages {
+			var gid int
+			for _, step := range stage.Steps {
+				pcounter++
+				if gid == 0 {
+					gid = pcounter
+				}
+				proc := &model.Proc{
+					BuildID: build.ID,
+					Name:    step.Alias,
+					PID:     pcounter,
+					PPID:    item.Proc.PID,
+					PGID:    gid,
+					State:   model.StatusPending,
+				}
+				if item.Proc.State == model.StatusSkipped {
+					proc.State = model.StatusSkipped
+				}
+				build.Procs = append(build.Procs, proc)
+			}
+		}
+	}
+}
+
 // return the metadata from the cli context.
 func metadataFromStruct(repo *model.Repo, build, last *model.Build, proc *model.Proc, link string) frontend.Metadata {
 	host := link
@@ -261,3 +305,10 @@ func metadataFromStruct(repo *model.Repo, build, last *model.Build, proc *model.
 		},
 	}
 }
+
+func sanitizePath(path string) string {
+	path = strings.TrimSuffix(path, ".yml")
+	path = strings.TrimPrefix(path, ".drone/")
+	path = strings.TrimPrefix(path, ".")
+	return path
+}

server/procBuilder_test.go

@@ -0,0 +1,206 @@
+// Copyright 2018 Drone.IO Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package server
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/laszlocph/drone-oss-08/model"
+	"github.com/laszlocph/drone-oss-08/remote"
+)
+
+func TestMultilineEnvsubst(t *testing.T) {
+	b := procBuilder{
+		Repo: &model.Repo{},
+		Curr: &model.Build{
+			Message: `aaa
+bbb`,
+		},
+		Last:  &model.Build{},
+		Netrc: &model.Netrc{},
+		Secs:  []*model.Secret{},
+		Regs:  []*model.Registry{},
+		Link:  "",
+		Yamls: []*remote.FileMeta{
+			&remote.FileMeta{Data: []byte(`
+pipeline:
+  xxx:
+    image: scratch
+    yyy: ${DRONE_COMMIT_MESSAGE}
+`)},
+			&remote.FileMeta{Data: []byte(`
+pipeline:
+  build:
+    image: scratch
+    yyy: ${DRONE_COMMIT_MESSAGE}
+`)},
+		}}
+
+	if buildItems, err := b.Build(); err != nil {
+		t.Fatal(err)
+	} else {
+		fmt.Println(buildItems)
+	}
+}
+
+func TestMultiPipeline(t *testing.T) {
+	b := procBuilder{
+		Repo:  &model.Repo{},
+		Curr:  &model.Build{},
+		Last:  &model.Build{},
+		Netrc: &model.Netrc{},
+		Secs:  []*model.Secret{},
+		Regs:  []*model.Registry{},
+		Link:  "",
+		Yamls: []*remote.FileMeta{
+			&remote.FileMeta{Data: []byte(`
+pipeline:
+  xxx:
+    image: scratch
+    yyy: ${DRONE_COMMIT_MESSAGE}
+`)},
+			&remote.FileMeta{Data: []byte(`
+pipeline:
+  build:
+    image: scratch
+    yyy: ${DRONE_COMMIT_MESSAGE}
+`)},
+		},
+	}
+
+	buildItems, err := b.Build()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(buildItems) != 2 {
+		t.Fatal("Should have generated 2 buildItems")
+	}
+}
+
+func TestDependsOn(t *testing.T) {
+	b := procBuilder{
+		Repo:  &model.Repo{},
+		Curr:  &model.Build{},
+		Last:  &model.Build{},
+		Netrc: &model.Netrc{},
+		Secs:  []*model.Secret{},
+		Regs:  []*model.Registry{},
+		Link:  "",
+		Yamls: []*remote.FileMeta{
+			&remote.FileMeta{Data: []byte(`
+pipeline:
+  deploy:
+    image: scratch
+
+depends_on:
+  - lint
+  - test
+  - build
+`)},
+		},
+	}
+
+	buildItems, err := b.Build()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(buildItems[0].DependsOn) != 3 {
+		t.Fatal("Should have 3 dependencies")
+	}
+	if buildItems[0].DependsOn[1] != "test" {
+		t.Fatal("Should depend on test")
+	}
+}
+
+func TestRunsOn(t *testing.T) {
+	b := procBuilder{
+		Repo:  &model.Repo{},
+		Curr:  &model.Build{},
+		Last:  &model.Build{},
+		Netrc: &model.Netrc{},
+		Secs:  []*model.Secret{},
+		Regs:  []*model.Registry{},
+		Link:  "",
+		Yamls: []*remote.FileMeta{
+			&remote.FileMeta{Data: []byte(`
+pipeline:
+  deploy:
+    image: scratch
+
+runs_on:
+  - success
+  - failure
+`)},
+		},
+	}
+
+	buildItems, err := b.Build()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(buildItems[0].RunsOn) != 2 {
+		t.Fatal("Should run on success and failure")
+	}
+	if buildItems[0].RunsOn[1] != "failure" {
+		t.Fatal("Should run on failure")
+	}
+}
+
+func TestBranchFilter(t *testing.T) {
+	b := procBuilder{
+		Repo:  &model.Repo{},
+		Curr:  &model.Build{Branch: "dev"},
+		Last:  &model.Build{},
+		Netrc: &model.Netrc{},
+		Secs:  []*model.Secret{},
+		Regs:  []*model.Registry{},
+		Link:  "",
+		Yamls: []*remote.FileMeta{
+			&remote.FileMeta{Data: []byte(`
+pipeline:
+  xxx:
+    image: scratch
+    yyy: ${DRONE_COMMIT_MESSAGE}
+branches: master
+`)},
+			&remote.FileMeta{Data: []byte(`
+pipeline:
+  build:
+    image: scratch
+    yyy: ${DRONE_COMMIT_MESSAGE}
+`)},
+		},
+	}
+
+	buildItems, err := b.Build()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(buildItems) != 2 {
+		t.Fatal("Should have generated 2 buildItems")
+	}
+	if buildItems[0].Proc.State != model.StatusSkipped {
+		t.Fatal("Should not run on dev branch")
+	}
+	for _, child := range buildItems[0].Proc.Children {
+		if child.State != model.StatusSkipped {
+			t.Fatal("Children should skipped status too")
+		}
+	}
+	if buildItems[1].Proc.State != model.StatusPending {
+		t.Fatal("Should not run on dev branch")
+	}
+}

server/repo.go

@@ -150,6 +150,9 @@ func PatchRepo(c *gin.Context) {
 	if in.BuildCounter != nil {
 		repo.Counter = *in.BuildCounter
 	}
+	if in.Fallback != nil {
+		repo.Fallback = *in.Fallback
+	}
 
 	err := store.UpdateRepo(c, repo)
 	if err != nil {

server/rpc.go

@@ -113,26 +113,22 @@ func (s *RPC) Next(c context.Context, filter rpc.Filter) (*rpc.Pipeline, error)
 	if err != nil {
 		return nil, err
 	}
-	task, err := s.queue.Poll(c, fn)
+	for {
-	if err != nil {
+		task, err := s.queue.Poll(c, fn)
-		return nil, err
+		if err != nil {
-	} else if task == nil {
+			return nil, err
-		return nil, nil
+		} else if task == nil {
+			return nil, nil
+		}
+
+		if task.ShouldRun() {
+			pipeline := new(rpc.Pipeline)
+			err = json.Unmarshal(task.Data, pipeline)
+			return pipeline, err
+		} else {
+			s.Done(c, task.ID, rpc.State{})
+		}
 	}
-	pipeline := new(rpc.Pipeline)
-
-	// check if the process was previously cancelled
-	// cancelled, _ := s.checkCancelled(pipeline)
-	// if cancelled {
-	// 	logrus.Debugf("ignore pid %v: cancelled by user", pipeline.ID)
-	// 	if derr := s.queue.Done(c, pipeline.ID); derr != nil {
-	// 		logrus.Errorf("error: done: cannot ack proc_id %v: %s", pipeline.ID, err)
-	// 	}
-	// 	return nil, nil
-	// }
-
-	err = json.Unmarshal(task.Data, pipeline)
-	return pipeline, err
 }
 
 // Wait implements the rpc.Wait function
@@ -383,76 +379,139 @@ func (s *RPC) Done(c context.Context, id string, state rpc.State) error {
 		return err
 	}
 
-	proc.Stopped = state.Finished
+	s.updateProcState(proc, state)
-	proc.Error = state.Error
-	proc.ExitCode = state.ExitCode
-	proc.State = model.StatusSuccess
-	if proc.ExitCode != 0 || proc.Error != "" {
-		proc.State = model.StatusFailure
-	}
-	if err := s.store.ProcUpdate(proc); err != nil {
-		log.Printf("error: done: cannot update proc_id %d state: %s", procID, err)
-	}
 
-	if err := s.queue.Done(c, id); err != nil {
+	var queueErr error
+	if proc.Failing() {
+		queueErr = s.queue.Error(c, id, fmt.Errorf("Proc finished with exitcode %d, %s", state.ExitCode, state.Error))
+	} else {
+		queueErr = s.queue.Done(c, id)
+	}
+	if queueErr != nil {
 		log.Printf("error: done: cannot ack proc_id %d: %s", procID, err)
 	}
 
-	// TODO handle this error
 	procs, _ := s.store.ProcList(build)
-	for _, p := range procs {
+	s.completeChildrenIfParentCompleted(procs, proc)
-		if p.Running() && p.PPID == proc.PID {
-			p.State = model.StatusSkipped
-			if p.Started != 0 {
-				p.State = model.StatusSuccess // for deamons that are killed
-				p.Stopped = proc.Stopped
-			}
-			if err := s.store.ProcUpdate(p); err != nil {
-				log.Printf("error: done: cannot update proc_id %d child state: %s", p.ID, err)
-			}
-		}
-	}
 
-	running := false
+	if !isThereRunningStage(procs) {
-	status := model.StatusSuccess
+		build.Status = buildStatus(procs)
-	for _, p := range procs {
-		if p.PPID == 0 {
-			if p.Running() {
-				running = true
-			}
-			if p.Failing() {
-				status = p.State
-			}
-		}
-	}
-	if !running {
-		build.Status = status
 		build.Finished = proc.Stopped
 		if err := s.store.UpdateBuild(build); err != nil {
 			log.Printf("error: done: cannot update build_id %d final state: %s", build.ID, err)
 		}
 
-		// update the status
+		if !isMultiPipeline(procs) {
-		user, err := s.store.GetUser(repo.UserID)
+			s.updateRemoteStatus(repo, build, nil)
-		if err == nil {
-			if refresher, ok := s.remote.(remote.Refresher); ok {
-				ok, _ := refresher.Refresh(user)
-				if ok {
-					s.store.UpdateUser(user)
-				}
-			}
-			uri := fmt.Sprintf("%s/%s/%d", s.host, repo.FullName, build.Number)
-			err = s.remote.Status(user, repo, build, uri)
-			if err != nil {
-				logrus.Errorf("error setting commit status for %s/%d: %v", repo.FullName, build.Number, err)
-			}
 		}
 	}
 
+	if isMultiPipeline(procs) {
+		s.updateRemoteStatus(repo, build, proc)
+	}
+
 	if err := s.logger.Close(c, id); err != nil {
 		log.Printf("error: done: cannot close build_id %d logger: %s", proc.ID, err)
 	}
 
+	s.notify(c, repo, build, procs)
+
+	return nil
+}
+
+func isMultiPipeline(procs []*model.Proc) bool {
+	countPPIDZero := 0
+	for _, proc := range procs {
+		if proc.PPID == 0 {
+			countPPIDZero++
+		}
+	}
+	return countPPIDZero > 1
+}
+
+// Log implements the rpc.Log function
+func (s *RPC) Log(c context.Context, id string, line *rpc.Line) error {
+	entry := new(logging.Entry)
+	entry.Data, _ = json.Marshal(line)
+	s.logger.Write(c, id, entry)
+	return nil
+}
+
+func (s *RPC) updateProcState(proc *model.Proc, state rpc.State) {
+	proc.Stopped = state.Finished
+	proc.Error = state.Error
+	proc.ExitCode = state.ExitCode
+	if state.Started == 0 {
+		proc.State = model.StatusSkipped
+	} else {
+		proc.State = model.StatusSuccess
+	}
+	if proc.ExitCode != 0 || proc.Error != "" {
+		proc.State = model.StatusFailure
+	}
+	if err := s.store.ProcUpdate(proc); err != nil {
+		log.Printf("error: done: cannot update proc_id %d state: %s", proc.ID, err)
+	}
+}
+
+func (s *RPC) completeChildrenIfParentCompleted(procs []*model.Proc, completedProc *model.Proc) {
+	for _, p := range procs {
+		if p.Running() && p.PPID == completedProc.PID {
+			p.State = model.StatusSkipped
+			if p.Started != 0 {
+				p.State = model.StatusSuccess // for deamons that are killed
+				p.Stopped = completedProc.Stopped
+			}
+			if err := s.store.ProcUpdate(p); err != nil {
+				log.Printf("error: done: cannot update proc_id %d child state: %s", p.ID, err)
+			}
+		}
+	}
+}
+
+func isThereRunningStage(procs []*model.Proc) bool {
+	for _, p := range procs {
+		if p.PPID == 0 {
+			if p.Running() {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+func buildStatus(procs []*model.Proc) string {
+	status := model.StatusSuccess
+
+	for _, p := range procs {
+		if p.PPID == 0 {
+			if p.Failing() {
+				status = p.State
+			}
+		}
+	}
+
+	return status
+}
+
+func (s *RPC) updateRemoteStatus(repo *model.Repo, build *model.Build, proc *model.Proc) {
+	user, err := s.store.GetUser(repo.UserID)
+	if err == nil {
+		if refresher, ok := s.remote.(remote.Refresher); ok {
+			ok, _ := refresher.Refresh(user)
+			if ok {
+				s.store.UpdateUser(user)
+			}
+		}
+		uri := fmt.Sprintf("%s/%s/%d", s.host, repo.FullName, build.Number)
+		err = s.remote.Status(user, repo, build, uri, proc)
+		if err != nil {
+			logrus.Errorf("error setting commit status for %s/%d: %v", repo.FullName, build.Number, err)
+		}
+	}
+}
+
+func (s *RPC) notify(c context.Context, repo *model.Repo, build *model.Build, procs []*model.Proc) {
 	build.Procs = model.Tree(procs)
 	message := pubsub.Message{
 		Labels: map[string]string{
@@ -465,31 +524,6 @@ func (s *RPC) Done(c context.Context, id string, state rpc.State) error {
 		Build: *build,
 	})
 	s.pubsub.Publish(c, "topic/events", message)
-
-	return nil
-}
-
-// Log implements the rpc.Log function
-func (s *RPC) Log(c context.Context, id string, line *rpc.Line) error {
-	entry := new(logging.Entry)
-	entry.Data, _ = json.Marshal(line)
-	s.logger.Write(c, id, entry)
-	return nil
-}
-
-func (s *RPC) checkCancelled(pipeline *rpc.Pipeline) (bool, error) {
-	pid, err := strconv.ParseInt(pipeline.ID, 10, 64)
-	if err != nil {
-		return false, err
-	}
-	proc, err := s.store.ProcLoad(pid)
-	if err != nil {
-		return false, err
-	}
-	if proc.State == model.StatusKilled {
-		return true, nil
-	}
-	return false, err
 }
 
 func createFilterFunc(filter rpc.Filter) (queue.Filter, error) {
@@ -561,42 +595,6 @@ func (s *DroneServer) Next(c oldcontext.Context, req *proto.NextRequest) (*proto
 	res.Pipeline.Payload, _ = json.Marshal(pipeline.Config)
 
 	return res, err
-
-	// fn := func(task *queue.Task) bool {
-	// 	for k, v := range req.GetFilter().Labels {
-	// 		if task.Labels[k] != v {
-	// 			return false
-	// 		}
-	// 	}
-	// 	return true
-	// }
-	// task, err := s.Queue.Poll(c, fn)
-	// if err != nil {
-	// 	return nil, err
-	// } else if task == nil {
-	// 	return nil, nil
-	// }
-	//
-	// pipeline := new(rpc.Pipeline)
-	// json.Unmarshal(task.Data, pipeline)
-	//
-	// res := new(proto.NextReply)
-	// res.Pipeline = new(proto.Pipeline)
-	// res.Pipeline.Id = pipeline.ID
-	// res.Pipeline.Timeout = pipeline.Timeout
-	// res.Pipeline.Payload, _ = json.Marshal(pipeline.Config)
-	//
-	// // check if the process was previously cancelled
-	// // cancelled, _ := s.checkCancelled(pipeline)
-	// // if cancelled {
-	// // 	logrus.Debugf("ignore pid %v: cancelled by user", pipeline.ID)
-	// // 	if derr := s.queue.Done(c, pipeline.ID); derr != nil {
-	// // 		logrus.Errorf("error: done: cannot ack proc_id %v: %s", pipeline.ID, err)
-	// // 	}
-	// // 	return nil, nil
-	// // }
-	//
-	// return res, nil
 }
 
 func (s *DroneServer) Init(c oldcontext.Context, req *proto.InitRequest) (*proto.Empty, error) {

store/datastore/builds_test.go

@@ -18,8 +18,8 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/laszlocph/drone-oss-08/model"
 	"github.com/franela/goblin"
+	"github.com/laszlocph/drone-oss-08/model"
 )
 
 func TestBuilds(t *testing.T) {

store/datastore/config.go

@@ -22,17 +22,17 @@ import (
 	"github.com/russross/meddler"
 )
 
-func (db *datastore) ConfigLoad(id int64) (*model.Config, error) {
+func (db *datastore) ConfigsForBuild(buildID int64) ([]*model.Config, error) {
 	stmt := sql.Lookup(db.driver, "config-find-id")
-	conf := new(model.Config)
+	var configs = []*model.Config{}
-	err := meddler.QueryRow(db, conf, stmt, id)
+	err := meddler.QueryAll(db, &configs, stmt, buildID)
-	return conf, err
+	return configs, err
 }
 
-func (db *datastore) ConfigFind(repo *model.Repo, hash string) (*model.Config, error) {
+func (db *datastore) ConfigFindIdentical(repoID int64, hash string) (*model.Config, error) {
 	stmt := sql.Lookup(db.driver, "config-find-repo-hash")
 	conf := new(model.Config)
-	err := meddler.QueryRow(db, conf, stmt, repo.ID, hash)
+	err := meddler.QueryRow(db, conf, stmt, repoID, hash)
 	return conf, err
 }
 
@@ -51,3 +51,7 @@ func (db *datastore) ConfigFindApproved(config *model.Config) (bool, error) {
 func (db *datastore) ConfigCreate(config *model.Config) error {
 	return meddler.Insert(db, "config", config)
 }
+
+func (db *datastore) BuildConfigCreate(buildConfig *model.BuildConfig) error {
+	return meddler.Insert(db, "build_config", buildConfig)
+}

store/datastore/config_test.go

@@ -23,6 +23,9 @@ import (
 func TestConfig(t *testing.T) {
 	s := newTest()
 	defer func() {
+		s.Exec("delete from repos")
+		s.Exec("delete from builds")
+		s.Exec("delete from procs")
 		s.Exec("delete from config")
 		s.Close()
 	}()
@@ -32,18 +35,49 @@ func TestConfig(t *testing.T) {
 		hash = "8d8647c9aa90d893bfb79dddbe901f03e258588121e5202632f8ae5738590b26"
 	)
 
-	if err := s.ConfigCreate(
+	repo := &model.Repo{
-		&model.Config{
+		UserID:   1,
-			RepoID: 2,
+		FullName: "bradrydzewski/drone",
-			Data:   data,
+		Owner:    "bradrydzewski",
-			Hash:   hash,
+		Name:     "drone",
-		},
+	}
-	); err != nil {
+	if err := s.CreateRepo(repo); err != nil {
+		t.Errorf("Unexpected error: insert repo: %s", err)
+		return
+	}
+
+	config := &model.Config{
+		RepoID: repo.ID,
+		Data:   data,
+		Hash:   hash,
+		Name:   "default",
+	}
+	if err := s.ConfigCreate(config); err != nil {
 		t.Errorf("Unexpected error: insert config: %s", err)
 		return
 	}
 
-	config, err := s.ConfigFind(&model.Repo{ID: 2}, hash)
+	build := &model.Build{
+		RepoID: repo.ID,
+		Status: model.StatusRunning,
+		Commit: "85f8c029b902ed9400bc600bac301a0aadb144ac",
+	}
+	if err := s.CreateBuild(build); err != nil {
+		t.Errorf("Unexpected error: insert build: %s", err)
+		return
+	}
+
+	if err := s.BuildConfigCreate(
+		&model.BuildConfig{
+			ConfigID: config.ID,
+			BuildID:  build.ID,
+		},
+	); err != nil {
+		t.Errorf("Unexpected error: insert build config: %s", err)
+		return
+	}
+
+	config, err := s.ConfigFindIdentical(repo.ID, hash)
 	if err != nil {
 		t.Error(err)
 		return
@@ -51,7 +85,7 @@ func TestConfig(t *testing.T) {
 	if got, want := config.ID, int64(1); got != want {
 		t.Errorf("Want config id %d, got %d", want, got)
 	}
-	if got, want := config.RepoID, int64(2); got != want {
+	if got, want := config.RepoID, repo.ID; got != want {
 		t.Errorf("Want config repo id %d, got %d", want, got)
 	}
 	if got, want := config.Data, data; got != want {
@@ -60,13 +94,16 @@ func TestConfig(t *testing.T) {
 	if got, want := config.Hash, hash; got != want {
 		t.Errorf("Want config hash %s, got %s", want, got)
 	}
+	if got, want := config.Name, "default"; got != want {
+		t.Errorf("Want config name %s, got %s", want, got)
+	}
 
-	loaded, err := s.ConfigLoad(config.ID)
+	loaded, err := s.ConfigsForBuild(build.ID)
 	if err != nil {
 		t.Errorf("Want config by id, got error %q", err)
 		return
 	}
-	if got, want := loaded.ID, config.ID; got != want {
+	if got, want := loaded[0].ID, config.ID; got != want {
 		t.Errorf("Want config by id %d, got %d", want, got)
 	}
 }
@@ -74,9 +111,10 @@ func TestConfig(t *testing.T) {
 func TestConfigApproved(t *testing.T) {
 	s := newTest()
 	defer func() {
-		s.Exec("delete from config")
-		s.Exec("delete from builds")
 		s.Exec("delete from repos")
+		s.Exec("delete from builds")
+		s.Exec("delete from procs")
+		s.Exec("delete from config")
 		s.Close()
 	}()
 
@@ -86,49 +124,83 @@ func TestConfigApproved(t *testing.T) {
 		Owner:    "bradrydzewski",
 		Name:     "drone",
 	}
-	s.CreateRepo(repo)
+	if err := s.CreateRepo(repo); err != nil {
+		t.Errorf("Unexpected error: insert repo: %s", err)
+		return
+	}
 
 	var (
-		data = "pipeline: [ { image: golang, commands: [ go build, go test ] } ]"
+		data         = "pipeline: [ { image: golang, commands: [ go build, go test ] } ]"
-		hash = "8d8647c9aa90d893bfb79dddbe901f03e258588121e5202632f8ae5738590b26"
+		hash         = "8d8647c9aa90d893bfb79dddbe901f03e258588121e5202632f8ae5738590b26"
-		conf = &model.Config{
+		buildBlocked = &model.Build{
 			RepoID: repo.ID,
-			Data:   data,
+			Status: model.StatusBlocked,
-			Hash:   hash,
+			Commit: "85f8c029b902ed9400bc600bac301a0aadb144ac",
+		}
+		buildPending = &model.Build{
+			RepoID: repo.ID,
+			Status: model.StatusPending,
+			Commit: "85f8c029b902ed9400bc600bac301a0aadb144ac",
+		}
+		buildRunning = &model.Build{
+			RepoID: repo.ID,
+			Status: model.StatusRunning,
+			Commit: "85f8c029b902ed9400bc600bac301a0aadb144ac",
 		}
 	)
 
+	if err := s.CreateBuild(buildBlocked); err != nil {
+		t.Errorf("Unexpected error: insert build: %s", err)
+		return
+	}
+	if err := s.CreateBuild(buildPending); err != nil {
+		t.Errorf("Unexpected error: insert build: %s", err)
+		return
+	}
+	conf := &model.Config{
+		RepoID: repo.ID,
+		Data:   data,
+		Hash:   hash,
+	}
 	if err := s.ConfigCreate(conf); err != nil {
 		t.Errorf("Unexpected error: insert config: %s", err)
 		return
 	}
-	s.CreateBuild(&model.Build{
+	buildConfig := &model.BuildConfig{
-		RepoID:   repo.ID,
 		ConfigID: conf.ID,
-		Status:   model.StatusBlocked,
+		BuildID:  buildBlocked.ID,
-		Commit:   "85f8c029b902ed9400bc600bac301a0aadb144ac",
+	}
-	})
+	if err := s.BuildConfigCreate(buildConfig); err != nil {
-	s.CreateBuild(&model.Build{
+		t.Errorf("Unexpected error: insert build_config: %s", err)
-		RepoID:   repo.ID,
-		ConfigID: conf.ID,
-		Status:   model.StatusPending,
-		Commit:   "85f8c029b902ed9400bc600bac301a0aadb144ac",
-	})
-
-	if ok, _ := s.ConfigFindApproved(conf); ok == true {
-		t.Errorf("Want config not approved, when blocked or pending")
 		return
 	}
 
-	s.CreateBuild(&model.Build{
+	if approved, err := s.ConfigFindApproved(conf); approved != false || err != nil {
-		RepoID:   repo.ID,
+		t.Errorf("Want config not approved, when blocked or pending. %v", err)
-		ConfigID: conf.ID,
+		return
-		Status:   model.StatusRunning,
+	}
-		Commit:   "85f8c029b902ed9400bc600bac301a0aadb144ac",
-	})
 
-	if ok, _ := s.ConfigFindApproved(conf); ok == false {
+	s.CreateBuild(buildRunning)
-		t.Errorf("Want config approved, when running.")
+	conf2 := &model.Config{
+		RepoID: repo.ID,
+		Data:   data,
+		Hash:   "xxx",
+	}
+	if err := s.ConfigCreate(conf2); err != nil {
+		t.Errorf("Unexpected error: insert config: %s", err)
+		return
+	}
+	buildConfig2 := &model.BuildConfig{
+		ConfigID: conf2.ID,
+		BuildID:  buildRunning.ID,
+	}
+	if err := s.BuildConfigCreate(buildConfig2); err != nil {
+		t.Errorf("Unexpected error: insert config: %s", err)
+		return
+	}
+
+	if approved, err := s.ConfigFindApproved(conf2); approved != true || err != nil {
+		t.Errorf("Want config approved, when running. %v", err)
 		return
 	}
 }
@@ -136,6 +208,9 @@ func TestConfigApproved(t *testing.T) {
 func TestConfigIndexes(t *testing.T) {
 	s := newTest()
 	defer func() {
+		s.Exec("delete from repos")
+		s.Exec("delete from builds")
+		s.Exec("delete from procs")
 		s.Exec("delete from config")
 		s.Close()
 	}()

store/datastore/ddl/mysql/ddl_gen.go

@@ -1,17 +1,3 @@
-// Copyright 2018 Drone.IO Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
 package mysql
 
 import (
@@ -170,6 +156,38 @@ var migrations = []struct {
 		name: "alter-table-update-file-meta",
 		stmt: alterTableUpdateFileMeta,
 	},
+	{
+		name: "create-table-build-config",
+		stmt: createTableBuildConfig,
+	},
+	{
+		name: "alter-table-add-config-name",
+		stmt: alterTableAddConfigName,
+	},
+	{
+		name: "update-table-set-config-name",
+		stmt: updateTableSetConfigName,
+	},
+	{
+		name: "populate-build-config",
+		stmt: populateBuildConfig,
+	},
+	{
+		name: "alter-table-add-task-dependencies",
+		stmt: alterTableAddTaskDependencies,
+	},
+	{
+		name: "alter-table-add-task-run-on",
+		stmt: alterTableAddTaskRunOn,
+	},
+	{
+		name: "alter-table-add-repo-fallback",
+		stmt: alterTableAddRepoFallback,
+	},
+	{
+		name: "update-table-set-repo-fallback",
+		stmt: updateTableSetRepoFallback,
+	},
 }
 
 // Migrate performs the database migration. If the migration fails
@@ -636,3 +654,62 @@ UPDATE files SET
 ,file_meta_failed=0
 ,file_meta_skipped=0
 `
+
+//
+// 019_create_table_build_config.sql
+//
+
+var createTableBuildConfig = `
+CREATE TABLE IF NOT EXISTS build_config (
+ config_id       INTEGER NOT NULL
+,build_id        INTEGER NOT NULL
+,PRIMARY KEY (config_id, build_id)
+,FOREIGN KEY (config_id) REFERENCES config (config_id)
+,FOREIGN KEY (build_id) REFERENCES builds (build_id)
+);
+`
+
+//
+// 020_add_column_config_name.sql
+//
+
+var alterTableAddConfigName = `
+ALTER TABLE config ADD COLUMN config_name TEXT
+`
+
+var updateTableSetConfigName = `
+UPDATE config SET config_name = "drone"
+`
+
+//
+// 021_populate_build_config.sql
+//
+
+var populateBuildConfig = `
+INSERT INTO build_config (config_id, build_id)
+SELECT build_config_id, build_id FROM builds
+`
+
+//
+// 022_add_task_columns.sql
+//
+
+var alterTableAddTaskDependencies = `
+ALTER TABLE tasks ADD COLUMN task_dependencies MEDIUMBLOB
+`
+
+var alterTableAddTaskRunOn = `
+ALTER TABLE tasks ADD COLUMN task_run_on MEDIUMBLOB
+`
+
+//
+// 023_add_repo_fallback_column.sql
+//
+
+var alterTableAddRepoFallback = `
+ALTER TABLE repos ADD COLUMN repo_fallback BOOLEAN
+`
+
+var updateTableSetRepoFallback = `
+UPDATE repos SET repo_fallback='false'
+`

store/datastore/ddl/mysql/files/019_create_table_build_config.sql

@@ -0,0 +1,9 @@
+-- name: create-table-build-config
+
+CREATE TABLE IF NOT EXISTS build_config (
+ config_id       INTEGER NOT NULL
+,build_id        INTEGER NOT NULL
+,PRIMARY KEY (config_id, build_id)
+,FOREIGN KEY (config_id) REFERENCES config (config_id)
+,FOREIGN KEY (build_id) REFERENCES builds (build_id)
+);

store/datastore/ddl/mysql/files/020_add_column_config_name.sql

@@ -0,0 +1,7 @@
+-- name: alter-table-add-config-name
+
+ALTER TABLE config ADD COLUMN config_name TEXT
+
+-- name: update-table-set-config-name
+
+UPDATE config SET config_name = "drone"

store/datastore/ddl/mysql/files/021_populate_build_config.sql

@@ -0,0 +1,4 @@
+-- name: populate-build-config
+
+INSERT INTO build_config (config_id, build_id)
+SELECT build_config_id, build_id FROM builds

store/datastore/ddl/mysql/files/022_add_task_columns.sql

@@ -0,0 +1,6 @@
+-- name: alter-table-add-task-dependencies
+ALTER TABLE tasks ADD COLUMN task_dependencies MEDIUMBLOB
+
+-- name: alter-table-add-task-run-on
+
+ALTER TABLE tasks ADD COLUMN task_run_on MEDIUMBLOB

store/datastore/ddl/mysql/files/023_add_repo_fallback_column.sql

@@ -0,0 +1,5 @@
+-- name: alter-table-add-repo-fallback
+ALTER TABLE repos ADD COLUMN repo_fallback BOOLEAN
+
+-- name: update-table-set-repo-fallback
+UPDATE repos SET repo_fallback='false'

store/datastore/ddl/postgres/ddl_gen.go

@@ -1,17 +1,3 @@
-// Copyright 2018 Drone.IO Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
 package postgres
 
 import (
@@ -170,6 +156,38 @@ var migrations = []struct {
 		name: "alter-table-update-file-meta",
 		stmt: alterTableUpdateFileMeta,
 	},
+	{
+		name: "create-table-build-config",
+		stmt: createTableBuildConfig,
+	},
+	{
+		name: "alter-table-add-config-name",
+		stmt: alterTableAddConfigName,
+	},
+	{
+		name: "update-table-set-config-name",
+		stmt: updateTableSetConfigName,
+	},
+	{
+		name: "populate-build-config",
+		stmt: populateBuildConfig,
+	},
+	{
+		name: "alter-table-add-task-dependencies",
+		stmt: alterTableAddTaskDependencies,
+	},
+	{
+		name: "alter-table-add-task-run-on",
+		stmt: alterTableAddTaskRunOn,
+	},
+	{
+		name: "alter-table-add-repo-fallback",
+		stmt: alterTableAddRepoFallback,
+	},
+	{
+		name: "update-table-set-repo-fallback",
+		stmt: updateTableSetRepoFallback,
+	},
 }
 
 // Migrate performs the database migration. If the migration fails
@@ -530,7 +548,7 @@ CREATE INDEX IF NOT EXISTS sender_repo_ix ON senders (sender_repo_id);
 //
 
 var alterTableAddRepoVisibility = `
-ALTER TABLE repos ADD COLUMN repo_visibility VARCHAR(50)
+ALTER TABLE repos ADD COLUMN repo_visibility VARCHAR(50);
 `
 
 var updateTableSetRepoVisibility = `
@@ -538,7 +556,7 @@ UPDATE repos
 SET repo_visibility = (CASE
   WHEN repo_private = false THEN 'public'
   ELSE 'private'
-  END)
+  END);
 `
 
 //
@@ -554,12 +572,13 @@ UPDATE repos SET repo_counter = (
   SELECT max(build_number)
   FROM builds
   WHERE builds.build_repo_id = repos.repo_id
-)
+);
 `
 
 var updateTableSetRepoSeqDefault = `
 UPDATE repos SET repo_counter = 0
 WHERE repo_counter IS NULL
+;
 `
 
 //
@@ -567,11 +586,11 @@ WHERE repo_counter IS NULL
 //
 
 var alterTableAddRepoActive = `
-ALTER TABLE repos ADD COLUMN repo_active BOOLEAN
+ALTER TABLE repos ADD COLUMN repo_active BOOLEAN;
 `
 
 var updateTableSetRepoActive = `
-UPDATE repos SET repo_active = true
+UPDATE repos SET repo_active = true;
 `
 
 //
@@ -583,7 +602,7 @@ ALTER TABLE users ADD COLUMN user_synced INTEGER;
 `
 
 var updateTableSetUserSynced = `
-UPDATE users SET user_synced = 0
+UPDATE users SET user_synced = 0;
 `
 
 //
@@ -615,19 +634,19 @@ CREATE INDEX IF NOT EXISTS ix_perms_user ON perms (perm_user_id);
 //
 
 var alterTableAddFilePid = `
-ALTER TABLE files ADD COLUMN file_pid INTEGER
+ALTER TABLE files ADD COLUMN file_pid INTEGER;
 `
 
 var alterTableAddFileMetaPassed = `
-ALTER TABLE files ADD COLUMN file_meta_passed INTEGER
+ALTER TABLE files ADD COLUMN file_meta_passed INTEGER;
 `
 
 var alterTableAddFileMetaFailed = `
-ALTER TABLE files ADD COLUMN file_meta_failed INTEGER
+ALTER TABLE files ADD COLUMN file_meta_failed INTEGER;
 `
 
 var alterTableAddFileMetaSkipped = `
-ALTER TABLE files ADD COLUMN file_meta_skipped INTEGER
+ALTER TABLE files ADD COLUMN file_meta_skipped INTEGER;
 `
 
 var alterTableUpdateFileMeta = `
@@ -635,4 +654,64 @@ UPDATE files SET
  file_meta_passed=0
 ,file_meta_failed=0
 ,file_meta_skipped=0
+;
+`
+
+//
+// 019_create_table_build_config.sql
+//
+
+var createTableBuildConfig = `
+CREATE TABLE IF NOT EXISTS build_config (
+ config_id       INTEGER NOT NULL
+,build_id        INTEGER NOT NULL
+,PRIMARY KEY (config_id, build_id)
+,FOREIGN KEY (config_id) REFERENCES config (config_id)
+,FOREIGN KEY (build_id) REFERENCES builds (build_id)
+);
+`
+
+//
+// 020_add_column_config_name.sql
+//
+
+var alterTableAddConfigName = `
+ALTER TABLE config ADD COLUMN config_name TEXT
+`
+
+var updateTableSetConfigName = `
+UPDATE config SET config_name = 'drone'
+`
+
+//
+// 021_populate_build_config.sql
+//
+
+var populateBuildConfig = `
+INSERT INTO build_config (config_id, build_id)
+SELECT build_config_id, build_id FROM builds
+`
+
+//
+// 022_add_task_columns.sql
+//
+
+var alterTableAddTaskDependencies = `
+ALTER TABLE tasks ADD COLUMN task_dependencies BYTEA
+`
+
+var alterTableAddTaskRunOn = `
+ALTER TABLE tasks ADD COLUMN task_run_on BYTEA
+`
+
+//
+// 023_add_repo_fallback_column.sql
+//
+
+var alterTableAddRepoFallback = `
+ALTER TABLE repos ADD COLUMN repo_fallback BOOLEAN
+`
+
+var updateTableSetRepoFallback = `
+UPDATE repos SET repo_fallback='false'
 `

store/datastore/ddl/postgres/files/019_create_table_build_config.sql

@@ -0,0 +1,9 @@
+-- name: create-table-build-config
+
+CREATE TABLE IF NOT EXISTS build_config (
+ config_id       INTEGER NOT NULL
+,build_id        INTEGER NOT NULL
+,PRIMARY KEY (config_id, build_id)
+,FOREIGN KEY (config_id) REFERENCES config (config_id)
+,FOREIGN KEY (build_id) REFERENCES builds (build_id)
+);

store/datastore/ddl/postgres/files/020_add_column_config_name.sql

@@ -0,0 +1,7 @@
+-- name: alter-table-add-config-name
+
+ALTER TABLE config ADD COLUMN config_name TEXT
+
+-- name: update-table-set-config-name
+
+UPDATE config SET config_name = 'drone'

store/datastore/ddl/postgres/files/021_populate_build_config.sql

@@ -0,0 +1,4 @@
+-- name: populate-build-config
+
+INSERT INTO build_config (config_id, build_id)
+SELECT build_config_id, build_id FROM builds

store/datastore/ddl/postgres/files/022_add_task_columns.sql

@@ -0,0 +1,6 @@
+-- name: alter-table-add-task-dependencies
+ALTER TABLE tasks ADD COLUMN task_dependencies BYTEA
+
+-- name: alter-table-add-task-run-on
+
+ALTER TABLE tasks ADD COLUMN task_run_on BYTEA

store/datastore/ddl/postgres/files/023_add_repo_fallback_column.sql

@@ -0,0 +1,5 @@
+-- name: alter-table-add-repo-fallback
+ALTER TABLE repos ADD COLUMN repo_fallback BOOLEAN
+
+-- name: update-table-set-repo-fallback
+UPDATE repos SET repo_fallback='false'

store/datastore/ddl/sqlite/ddl_gen.go

@@ -1,17 +1,3 @@
-// Copyright 2018 Drone.IO Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
 package sqlite
 
 import (
@@ -174,6 +160,38 @@ var migrations = []struct {
 		name: "alter-table-update-file-meta",
 		stmt: alterTableUpdateFileMeta,
 	},
+	{
+		name: "create-table-build-config",
+		stmt: createTableBuildConfig,
+	},
+	{
+		name: "alter-table-add-config-name",
+		stmt: alterTableAddConfigName,
+	},
+	{
+		name: "update-table-set-config-name",
+		stmt: updateTableSetConfigName,
+	},
+	{
+		name: "populate-build-config",
+		stmt: populateBuildConfig,
+	},
+	{
+		name: "alter-table-add-task-dependencies",
+		stmt: alterTableAddTaskDependencies,
+	},
+	{
+		name: "alter-table-add-task-run-on",
+		stmt: alterTableAddTaskRunOn,
+	},
+	{
+		name: "alter-table-add-repo-fallback",
+		stmt: alterTableAddRepoFallback,
+	},
+	{
+		name: "update-table-set-repo-fallback",
+		stmt: updateTableSetRepoFallback,
+	},
 }
 
 // Migrate performs the database migration. If the migration fails
@@ -637,3 +655,62 @@ UPDATE files SET
 ,file_meta_failed=0
 ,file_meta_skipped=0
 `
+
+//
+// 019_create_table_build_config.sql
+//
+
+var createTableBuildConfig = `
+CREATE TABLE IF NOT EXISTS build_config (
+ config_id       INTEGER NOT NULL
+,build_id        INTEGER NOT NULL
+,PRIMARY KEY (config_id, build_id)
+,FOREIGN KEY (config_id) REFERENCES config (config_id)
+,FOREIGN KEY (build_id) REFERENCES builds (build_id)
+);
+`
+
+//
+// 020_add_column_config_name.sql
+//
+
+var alterTableAddConfigName = `
+ALTER TABLE config ADD COLUMN config_name TEXT
+`
+
+var updateTableSetConfigName = `
+UPDATE config SET config_name = "drone"
+`
+
+//
+// 021_populate_build_config.sql
+//
+
+var populateBuildConfig = `
+INSERT INTO build_config (config_id, build_id)
+SELECT build_config_id, build_id FROM builds
+`
+
+//
+// 022_add_task_columns.sql
+//
+
+var alterTableAddTaskDependencies = `
+ALTER TABLE tasks ADD COLUMN task_dependencies BLOB
+`
+
+var alterTableAddTaskRunOn = `
+ALTER TABLE tasks ADD COLUMN task_run_on BLOB
+`
+
+//
+// 023_add_repo_fallback_column.sql
+//
+
+var alterTableAddRepoFallback = `
+ALTER TABLE repos ADD COLUMN repo_fallback BOOLEAN
+`
+
+var updateTableSetRepoFallback = `
+UPDATE repos SET repo_fallback='false'
+`

store/datastore/ddl/sqlite/files/019_create_table_build_config.sql

@@ -0,0 +1,9 @@
+-- name: create-table-build-config
+
+CREATE TABLE IF NOT EXISTS build_config (
+ config_id       INTEGER NOT NULL
+,build_id        INTEGER NOT NULL
+,PRIMARY KEY (config_id, build_id)
+,FOREIGN KEY (config_id) REFERENCES config (config_id)
+,FOREIGN KEY (build_id) REFERENCES builds (build_id)
+);

store/datastore/ddl/sqlite/files/020_add_column_config_name.sql

@@ -0,0 +1,7 @@
+-- name: alter-table-add-config-name
+
+ALTER TABLE config ADD COLUMN config_name TEXT
+
+-- name: update-table-set-config-name
+
+UPDATE config SET config_name = "drone"

store/datastore/ddl/sqlite/files/021_populate_build_config.sql

@@ -0,0 +1,4 @@
+-- name: populate-build-config
+
+INSERT INTO build_config (config_id, build_id)
+SELECT build_config_id, build_id FROM builds

store/datastore/ddl/sqlite/files/022_add_task_columns.sql

@@ -0,0 +1,6 @@
+-- name: alter-table-add-task-dependencies
+ALTER TABLE tasks ADD COLUMN task_dependencies BLOB
+
+-- name: alter-table-add-task-run-on
+
+ALTER TABLE tasks ADD COLUMN task_run_on BLOB

store/datastore/ddl/sqlite/files/023_add_repo_fallback_column.sql

@@ -0,0 +1,5 @@
+-- name: alter-table-add-repo-fallback
+ALTER TABLE repos ADD COLUMN repo_fallback BOOLEAN
+
+-- name: update-table-set-repo-fallback
+UPDATE repos SET repo_fallback='false'

store/datastore/sql/mysql/files/config.sql

@@ -1,12 +1,14 @@
 -- name: config-find-id
 
 SELECT
- config_id
+ config.config_id
 ,config_repo_id
 ,config_hash
 ,config_data
+,config_name
 FROM config
-WHERE config_id = ?
+LEFT JOIN build_config ON config.config_id = build_config.config_id
+WHERE build_config.build_id = ?
 
 -- name: config-find-repo-hash
 
@@ -15,6 +17,7 @@ SELECT
 ,config_repo_id
 ,config_hash
 ,config_data
+,config_name
 FROM config
 WHERE config_repo_id = ?
   AND config_hash    = ?
@@ -23,6 +26,10 @@ WHERE config_repo_id = ?
 
 SELECT build_id FROM builds
 WHERE build_repo_id = ?
-AND build_config_id = ?
+AND build_id in (
+  SELECT build_id
+  FROM build_config
+  WHERE build_config.config_id = ?
+  )
 AND build_status NOT IN ('blocked', 'pending')
 LIMIT 1

store/datastore/sql/mysql/files/task.sql

@@ -4,6 +4,8 @@ SELECT
  task_id
 ,task_data
 ,task_labels
+,task_dependencies
+,task_run_on
 FROM tasks
 
 -- name: task-delete

store/datastore/sql/mysql/sql_gen.go

@@ -55,12 +55,14 @@ var index = map[string]string{
 
 var configFindId = `
 SELECT
- config_id
+ config.config_id
 ,config_repo_id
 ,config_hash
 ,config_data
+,config_name
 FROM config
-WHERE config_id = ?
+LEFT JOIN build_config ON config.config_id = build_config.config_id
+WHERE build_config.build_id = ?
 `
 
 var configFindRepoHash = `
@@ -69,6 +71,7 @@ SELECT
 ,config_repo_id
 ,config_hash
 ,config_data
+,config_name
 FROM config
 WHERE config_repo_id = ?
   AND config_hash    = ?
@@ -77,7 +80,11 @@ WHERE config_repo_id = ?
 var configFindApproved = `
 SELECT build_id FROM builds
 WHERE build_repo_id = ?
-AND build_config_id = ?
+AND build_id in (
+  SELECT build_id
+  FROM build_config
+  WHERE build_config.config_id = ?
+  )
 AND build_status NOT IN ('blocked', 'pending')
 LIMIT 1
 `
@@ -547,6 +554,8 @@ SELECT
  task_id
 ,task_data
 ,task_labels
+,task_dependencies
+,task_run_on
 FROM tasks
 `
 

store/datastore/sql/postgres/files/config.sql

@@ -1,12 +1,14 @@
 -- name: config-find-id
 
 SELECT
- config_id
+ config.config_id
 ,config_repo_id
 ,config_hash
 ,config_data
+,config_name
 FROM config
-WHERE config_id = $1
+LEFT JOIN build_config ON config.config_id = build_config.config_id
+WHERE build_config.build_id = $1
 
 -- name: config-find-repo-hash
 
@@ -15,6 +17,7 @@ SELECT
 ,config_repo_id
 ,config_hash
 ,config_data
+,config_name
 FROM config
 WHERE config_repo_id = $1
   AND config_hash    = $2
@@ -23,6 +26,10 @@ WHERE config_repo_id = $1
 
 SELECT build_id FROM builds
 WHERE build_repo_id = $1
-AND build_config_id = $2
+AND build_id in (
+  SELECT build_id
+  FROM build_config
+  WHERE build_config.config_id = $2
+  )
 AND build_status NOT IN ('blocked', 'pending')
 LIMIT 1

store/datastore/sql/postgres/files/tasks.sql

@@ -4,6 +4,8 @@ SELECT
  task_id
 ,task_data
 ,task_labels
+,task_dependencies
+,task_run_on
 FROM tasks
 
 -- name: task-delete

store/datastore/sql/postgres/sql_gen.go

@@ -55,12 +55,14 @@ var index = map[string]string{
 
 var configFindId = `
 SELECT
- config_id
+ config.config_id
 ,config_repo_id
 ,config_hash
 ,config_data
+,config_name
 FROM config
-WHERE config_id = $1
+LEFT JOIN build_config ON config.config_id = build_config.config_id
+WHERE build_config.build_id = $1
 `
 
 var configFindRepoHash = `
@@ -69,6 +71,7 @@ SELECT
 ,config_repo_id
 ,config_hash
 ,config_data
+,config_name
 FROM config
 WHERE config_repo_id = $1
   AND config_hash    = $2
@@ -77,7 +80,11 @@ WHERE config_repo_id = $1
 var configFindApproved = `
 SELECT build_id FROM builds
 WHERE build_repo_id = $1
-AND build_config_id = $2
+AND build_id in (
+  SELECT build_id
+  FROM build_config
+  WHERE build_config.config_id = $2
+  )
 AND build_status NOT IN ('blocked', 'pending')
 LIMIT 1
 `
@@ -95,7 +102,7 @@ WHERE repo_active = true
 
 var countBuilds = `
 SELECT count(1)
-FROM builds;
+FROM builds
 `
 
 var feedLatestBuild = `
@@ -552,6 +559,8 @@ SELECT
  task_id
 ,task_data
 ,task_labels
+,task_dependencies
+,task_run_on
 FROM tasks
 `
 

store/datastore/sql/sqlite/files/config.sql

@@ -1,12 +1,14 @@
 -- name: config-find-id
 
 SELECT
- config_id
+ config.config_id
 ,config_repo_id
 ,config_hash
 ,config_data
+,config_name
 FROM config
-WHERE config_id = ?
+LEFT JOIN build_config ON config.config_id = build_config.config_id
+WHERE build_config.build_id = ?
 
 -- name: config-find-repo-hash
 
@@ -15,6 +17,7 @@ SELECT
 ,config_repo_id
 ,config_hash
 ,config_data
+,config_name
 FROM config
 WHERE config_repo_id = ?
   AND config_hash    = ?
@@ -23,6 +26,10 @@ WHERE config_repo_id = ?
 
 SELECT build_id FROM builds
 WHERE build_repo_id = ?
-AND build_config_id = ?
+AND build_id in (
+  SELECT build_id
+  FROM build_config
+  WHERE build_config.config_id = ?
+  )
 AND build_status NOT IN ('blocked', 'pending')
 LIMIT 1

store/datastore/sql/sqlite/files/task.sql

@@ -4,6 +4,8 @@ SELECT
  task_id
 ,task_data
 ,task_labels
+,task_dependencies
+,task_run_on
 FROM tasks
 
 -- name: task-delete

store/datastore/sql/sqlite/sql_gen.go

@@ -55,12 +55,14 @@ var index = map[string]string{
 
 var configFindId = `
 SELECT
- config_id
+ config.config_id
 ,config_repo_id
 ,config_hash
 ,config_data
+,config_name
 FROM config
-WHERE config_id = ?
+LEFT JOIN build_config ON config.config_id = build_config.config_id
+WHERE build_config.build_id = ?
 `
 
 var configFindRepoHash = `
@@ -69,6 +71,7 @@ SELECT
 ,config_repo_id
 ,config_hash
 ,config_data
+,config_name
 FROM config
 WHERE config_repo_id = ?
   AND config_hash    = ?
@@ -77,7 +80,11 @@ WHERE config_repo_id = ?
 var configFindApproved = `
 SELECT build_id FROM builds
 WHERE build_repo_id = ?
-AND build_config_id = ?
+AND build_id in (
+  SELECT build_id
+  FROM build_config
+  WHERE build_config.config_id = ?
+  )
 AND build_status NOT IN ('blocked', 'pending')
 LIMIT 1
 `
@@ -547,6 +554,8 @@ SELECT
  task_id
 ,task_data
 ,task_labels
+,task_dependencies
+,task_run_on
 FROM tasks
 `
 

store/store.go

@@ -111,10 +111,11 @@ type Store interface {
 	PermDelete(perm *model.Perm) error
 	PermFlush(user *model.User, before int64) error
 
-	ConfigLoad(int64) (*model.Config, error)
+	ConfigsForBuild(buildID int64) ([]*model.Config, error)
-	ConfigFind(*model.Repo, string) (*model.Config, error)
+	ConfigFindIdentical(repoID int64, sha string) (*model.Config, error)
 	ConfigFindApproved(*model.Config) (bool, error)
 	ConfigCreate(*model.Config) error
+	BuildConfigCreate(*model.BuildConfig) error
 
 	SenderFind(*model.Repo, string) (*model.Sender, error)
 	SenderList(*model.Repo) ([]*model.Sender, error)

vendor/vendor.json

@@ -425,10 +425,12 @@
 			"revisionTime": "2016-05-04T02:26:26Z"
 		},
 		{
-			"checksumSHA1": "i3dVVpc0/It5nJIt/LEOHNuvqQY=",
+			"checksumSHA1": "R/gRUF6hXEFbDGSIOKt6VdCPwHE=",
 			"path": "github.com/laszlocph/drone-ui/dist",
-			"revision": "106788432c8e9f19ee7a73fd977ef15d32cca79d",
+			"revision": "1c55c6bab89440efc658a708ba7bb3424dbd5d0d",
-			"revisionTime": "2019-06-24T07:03:37Z"
+			"revisionTime": "2019-06-25T11:41:53Z",
+			"version": "fallback-config",
+			"versionExact": "fallback-config"
 		},
 		{
 			"path": "github.com/lib/pq",