<commit_before>package topgun_test

import (
	"bytes"
	"database/sql"
	"fmt"
	"io/ioutil"
	"regexp"
	"time"

	_ "github.com/lib/pq"
	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	"github.com/onsi/gomega/gbytes"
	"github.com/onsi/gomega/gexec"
)

var _ = Describe("[#129726011] Worker stalling", func() {
	var dbConn *sql.DB

	BeforeEach(func() {
		var err error
		dbConn, err = sql.Open("postgres", fmt.Sprintf("postgres://atc:dummy-password@%s:5432/atc?sslmode=disable", atcIP))
		Expect(err).ToNot(HaveOccurred())
	})

	waitForStalledWorker := func() string {
		var stalledWorkerName string
		Eventually(func() string {
			rows, err := psql.Select("name, state").From("workers").RunWith(dbConn).Query()
			Expect(err).ToNot(HaveOccurred())

			for rows.Next() {
				var name string
				var state string

				err := rows.Scan(&name, &state)
				Expect(err).ToNot(HaveOccurred())

				if state != "stalled" {
					continue
				}

				if stalledWorkerName != "" {
					Fail("multiple workers stalled")
				}

				stalledWorkerName = name
			}

			return stalledWorkerName
		}).ShouldNot(BeEmpty())

		return stalledWorkerName
	}

	waitForWorkersToBeRunning := func() {
		Eventually(func() bool {
			rows, err := psql.Select("name, state").From("workers").RunWith(dbConn).Query()
			Expect(err).ToNot(HaveOccurred())

			anyStalled := false

			for rows.Next() {
				var name string
				var state string

				err := rows.Scan(&name, &state)
				Expect(err).ToNot(HaveOccurred())

				if state == "stalled" {
					anyStalled = true
				}
			}

			return anyStalled
		}).Should(BeFalse())
	}

	Context("with two workers available", func() {
		BeforeEach(func() {
			Deploy("deployments/two-forwarded-workers.yml")
		})

		It("initially runs tasks across all workers", func() {
			Eventually(func() map[string]struct{} {
				fly("execute", "-c", "tasks/tiny.yml")
				rows, err := psql.Select("id, worker_name").From("containers").RunWith(dbConn).Query()
				Expect(err).ToNot(HaveOccurred())

				usedWorkers := map[string]struct{}{}
				for rows.Next() {
					var id int
					var workerName string
					err := rows.Scan(&id, &workerName)
					Expect(err).ToNot(HaveOccurred())
					usedWorkers[workerName] = struct{}{}
				}

				return usedWorkers
			}, 10*time.Minute).Should(HaveLen(2))
		})

		Context("when one worker goes away", func() {
			var stalledWorkerName string

			BeforeEach(func() {
				bosh("ssh", "worker/0", "-c", "sudo /var/vcap/bosh/bin/monit stop all")
				stalledWorkerName = waitForStalledWorker()
			})

			It("enters 'stalled' state and is no longer used for new containers", func() {
				for i := 0; i < 10; i++ {
					fly("execute", "-c", "tasks/tiny.yml")
					rows, err := psql.Select("id, worker_name").From("containers").RunWith(dbConn).Query()
					Expect(err).ToNot(HaveOccurred())

					usedWorkers := map[string]struct{}{}
					for rows.Next() {
						var id int
						var workerName string
						err := rows.Scan(&id, &workerName)
						Expect(err).ToNot(HaveOccurred())
						usedWorkers[workerName] = struct{}{}
					}

					Expect(usedWorkers).To(HaveLen(1))
					Expect(usedWorkers).ToNot(ContainElement(stalledWorkerName))
				}
			})
		})
	})

	Context("with no other worker available", func() {
		BeforeEach(func() {
			Deploy("deployments/single-vm-forwarded-worker.yml")
		})

		Context("when the worker stalls while a build is running", func() {
			var stalledWorkerName string

			var buildSession *gexec.Session
			var buildID string

			BeforeEach(func() {
				buildSession = spawnFly("execute", "-c", "tasks/wait.yml")
				Eventually(buildSession).Should(gbytes.Say("executing build"))

				buildRegex := regexp.MustCompile(`executing build (\d+)`)
				matches := buildRegex.FindSubmatch(buildSession.Out.Contents())
				buildID = string(matches[1])

				Eventually(buildSession).Should(gbytes.Say("waiting"))

				By("stopping the worker without draining")
				bosh("ssh", "concourse/0", "-c", "sudo /var/vcap/bosh/bin/monit stop all")

				By("waiting for it to stall")
				stalledWorkerName = waitForStalledWorker()
			})

			It("does not fail the build", func() {
				Consistently(buildSession).ShouldNot(gexec.Exit())
			})

			Context("when the worker comes back", func() {
				BeforeEach(func() {
					By("starting the worker")
					bosh("ssh", "concourse/0", "-c", "sudo /var/vcap/bosh/bin/monit start all")

					By("waiting for it to be running again")
					waitForWorkersToBeRunning()
				})

				It("resumes the build", func() {
					_, err := ioutil.ReadAll(buildSession.Out)
					Expect(err).ToNot(HaveOccurred())

					By("reattaching to the build")
					Eventually(buildSession).Should(gbytes.Say("waiting"))

					By("hijacking the build to tell it to finish")<commit_msg>woops, 'all' is a bit aggressive<commit_after>package topgun_test

import (
	"bytes"
	"database/sql"
	"fmt"
	"io/ioutil"
	"regexp"
	"time"

	_ "github.com/lib/pq"
	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	"github.com/onsi/gomega/gbytes"
	"github.com/onsi/gomega/gexec"
)

var _ = Describe("[#129726011] Worker stalling", func() {
	var dbConn *sql.DB

	BeforeEach(func() {
		var err error
		dbConn, err = sql.Open("postgres", fmt.Sprintf("postgres://atc:dummy-password@%s:5432/atc?sslmode=disable", atcIP))
		Expect(err).ToNot(HaveOccurred())
	})

	waitForStalledWorker := func() string {
		var stalledWorkerName string
		Eventually(func() string {
			rows, err := psql.Select("name, state").From("workers").RunWith(dbConn).Query()
			Expect(err).ToNot(HaveOccurred())

			for rows.Next() {
				var name string
				var state string

				err := rows.Scan(&name, &state)
				Expect(err).ToNot(HaveOccurred())

				if state != "stalled" {
					continue
				}

				if stalledWorkerName != "" {
					Fail("multiple workers stalled")
				}

				stalledWorkerName = name
			}

			return stalledWorkerName
		}).ShouldNot(BeEmpty())

		return stalledWorkerName
	}

	waitForWorkersToBeRunning := func() {
		Eventually(func() bool {
			rows, err := psql.Select("name, state").From("workers").RunWith(dbConn).Query()
			Expect(err).ToNot(HaveOccurred())

			anyStalled := false

			for rows.Next() {
				var name string
				var state string

				err := rows.Scan(&name, &state)
				Expect(err).ToNot(HaveOccurred())

				if state == "stalled" {
					anyStalled = true
				}
			}

			return anyStalled
		}).Should(BeFalse())
	}

	Context("with two workers available", func() {
		BeforeEach(func() {
			Deploy("deployments/two-forwarded-workers.yml")
		})

		It("initially runs tasks across all workers", func() {
			Eventually(func() map[string]struct{} {
				fly("execute", "-c", "tasks/tiny.yml")
				rows, err := psql.Select("id, worker_name").From("containers").RunWith(dbConn).Query()
				Expect(err).ToNot(HaveOccurred())

				usedWorkers := map[string]struct{}{}
				for rows.Next() {
					var id int
					var workerName string
					err := rows.Scan(&id, &workerName)
					Expect(err).ToNot(HaveOccurred())
					usedWorkers[workerName] = struct{}{}
				}

				return usedWorkers
			}, 10*time.Minute).Should(HaveLen(2))
		})

		Context("when one worker goes away", func() {
			var stalledWorkerName string

			BeforeEach(func() {
				bosh("ssh", "worker/0", "-c", "sudo /var/vcap/bosh/bin/monit stop beacon garden")
				stalledWorkerName = waitForStalledWorker()
			})

			It("enters 'stalled' state and is no longer used for new containers", func() {
				for i := 0; i < 10; i++ {
					fly("execute", "-c", "tasks/tiny.yml")
					rows, err := psql.Select("id, worker_name").From("containers").RunWith(dbConn).Query()
					Expect(err).ToNot(HaveOccurred())

					usedWorkers := map[string]struct{}{}
					for rows.Next() {
						var id int
						var workerName string
						err := rows.Scan(&id, &workerName)
						Expect(err).ToNot(HaveOccurred())
						usedWorkers[workerName] = struct{}{}
					}

					Expect(usedWorkers).To(HaveLen(1))
					Expect(usedWorkers).ToNot(ContainElement(stalledWorkerName))
				}
			})
		})
	})

	Context("with no other worker available", func() {
		BeforeEach(func() {
			Deploy("deployments/single-vm-forwarded-worker.yml")
		})

		Context("when the worker stalls while a build is running", func() {
			var stalledWorkerName string

			var buildSession *gexec.Session
			var buildID string

			BeforeEach(func() {
				buildSession = spawnFly("execute", "-c", "tasks/wait.yml")
				Eventually(buildSession).Should(gbytes.Say("executing build"))

				buildRegex := regexp.MustCompile(`executing build (\d+)`)
				matches := buildRegex.FindSubmatch(buildSession.Out.Contents())
				buildID = string(matches[1])

				Eventually(buildSession).Should(gbytes.Say("waiting"))

				By("stopping the worker without draining")
				bosh("ssh", "concourse/0", "-c", "sudo /var/vcap/bosh/bin/monit stop beacon garden")

				By("waiting for it to stall")
				stalledWorkerName = waitForStalledWorker()
			})

			It("does not fail the build", func() {
				Consistently(buildSession).ShouldNot(gexec.Exit())
			})

			Context("when the worker comes back", func() {
				BeforeEach(func() {
					By("starting the worker")
					bosh("ssh", "concourse/0", "-c", "sudo /var/vcap/bosh/bin/monit start beacon garden")

					By("waiting for it to be running again")
					waitForWorkersToBeRunning()
				})

				It("resumes the build", func() {
					_, err := ioutil.ReadAll(buildSession.Out)
					Expect(err).ToNot(HaveOccurred())

					By("reattaching to the build")
					Eventually(buildSession).Should(gbytes.Say("waiting"))

					By("hijacking the build to tell it to finish")
{'fraction_non_alphanumeric': 0.15767417496071243, 'fraction_numerical': 0.005028810895756941, 'mean_word_length': 3.6004819277108435, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3708423', 'n_tokens_mistral': 3976, 'n_tokens_neox': 3142, 'n_words': 655}