Sign in

Search issue titles and bodies, change requests and comments. Private repositories you cannot read are not included.

vole
Clone

Read only, no credentials:

https://rodent.software/vole.git

Write over SSH with a registered key:

git clone ssh://git@rodent.software:2222/vole.git

A free and open-source forge that lives in your Git repo.

Phase 4 slice 2: the queue and the runner

b44cbfaf873fda01a458b527e5684b2386266117 · Keenan Nemetz · 18d ago · parent 8fe238d4

Phase 4 slice 2: the queue and the runner

A push now runs a pipeline. post-receive spools one file per accepted ref update —
the hook must never write to a database or fail a commit — and the server ingests
those into data/runs.db, pinning each run to the commit and to the workflow blob it
was queued from. The unique index on (repo, ref, sha, workflow, job) is what makes
ingestion idempotent, so a spool file read twice after a crash does not run twice.

vole runner connects over SSH with the key it would push with: no token, no stored
secret, and the same identity the rest of the system uses. It may only run jobs for
repositories where its account is an admin, because running a job executes that
repository's code — a bigger power than writing to it.

Delivery is at-least-once by construction:

- A lease hands out a job for a fixed window; the runner heartbeats while it works.
- An expired lease means the runner died, and the janitor returns the job to the
  queue for somebody else.
- A runner whose lease is gone stops rather than reporting a result nobody will
  record, and a job that kills five runners in a row is failed instead of cycling.

Three bugs found while building it, each worth the test that now covers it:

- Background loops lived in Run(), so the tests — which serve the SSH listener
  themselves — never had ingestion or lease expiry. They start in New() now.
- post-receive derived the repository name from GIT_DIR, which git sets to "." for
  hooks, so every event was queued for a repository called ".". The same trap had
  already appeared in a pre-receive message; both now resolve the absolute path.
- Leases ordered only by creation time, so two pushes in the same second could be
  handed out in any order; the id breaks the tie, and ULIDs sort by time.

Verified end to end on a live instance: a push queues a job, a runner leases it,
clones the commit, prints "hi", and the run closes success with the log attached.
The integration tests cover the two exit criteria that matter — a push runs a
pipeline, and killing a runner mid-job requeues the work onto a second runner
(attempts=2, success) — plus the property that editing a workflow after a push does
not change what the already-queued run executes.

Not built yet: statuses on change requests and required-check gating (slice 3), and
signed run records (Phase 7). Runs live in data/runs.db and are visible there, not
on the web.

internal/server/integration_test.go +214−1

@@ -18,7 +18,9 @@ import (
18 18 "path/filepath"
19 19 "strings"
20 20 "sync"
21 "syscall"
21 22 "testing"
23 "time"
22 24
23 25 "golang.org/x/crypto/ssh"
24 26
@@ -27,6 +29,7 @@ import (
27 29 "github.com/rodent-software/vole/internal/identity"
28 30 "github.com/rodent-software/vole/internal/index"
29 31 "github.com/rodent-software/vole/internal/repo"
32 "github.com/rodent-software/vole/internal/runs"
30 33 "github.com/rodent-software/vole/internal/server"
31 34 )
32 35
@@ -63,13 +66,20 @@ type fixture struct {
63 66 writerID string
64 67 }
65 68
66 func newFixture(t *testing.T) *fixture {
69 func newFixture(t *testing.T) *fixture { return newFixtureWith(t, nil) }
70
71 // newFixtureWith allows a test to adjust configuration before the server starts,
72 // which is how the CI tests get a short lease.
73 func newFixtureWith(t *testing.T, tweak func(*config.Config)) *fixture {
67 74 t.Helper()
68 75 cfg, err := config.Load("")
69 76 if err != nil {
70 77 t.Fatal(err)
71 78 }
72 79 cfg.DataDir = t.TempDir()
80 if tweak != nil {
81 tweak(cfg)
82 }
73 83 if err := os.MkdirAll(cfg.ReposDir(), 0o755); err != nil {
74 84 t.Fatal(err)
75 85 }
@@ -1693,6 +1703,209 @@ func TestInvalidWorkflowRejectedAtPush(t *testing.T) {
1693 1703 }
1694 1704 }
1695 1705
1706 // ---------- CI ----------
1707
1708 // ciEnv is the environment a runner needs: the same key git would push with.
1709 func (f *fixture) ciEnv() []string {
1710 sshCmd := "ssh -i " + f.keyPath + " -o IdentitiesOnly=yes -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null -o LogLevel=ERROR"
1711 return append(os.Environ(), "VOLE_SSH_COMMAND="+sshCmd, "GIT_SSH_COMMAND="+sshCmd,
1712 "GIT_TERMINAL_PROMPT=0", "GIT_CONFIG_GLOBAL=/dev/null", "GIT_CONFIG_SYSTEM=/dev/null")
1713 }
1714
1715 func (f *fixture) sshBase() string { return "ssh://git@" + f.sshAddr }
1716
1717 // waitForRun polls the queue until a run matches, because ingestion is a ticker.
1718 func (f *fixture) waitForRun(t *testing.T, within time.Duration, match func(*runs.Run) bool) *runs.Run {
1719 t.Helper()
1720 deadline := time.Now().Add(within)
1721 var last []*runs.Run
1722 for time.Now().Before(deadline) {
1723 list, err := f.srv.Runs.Recent(f.repo, 20)
1724 if err == nil {
1725 last = list
1726 for _, r := range list {
1727 if match(r) {
1728 return r
1729 }
1730 }
1731 }
1732 time.Sleep(100 * time.Millisecond)
1733 }
1734 t.Fatalf("no matching run after %s; saw %d run(s): %+v", within, len(last), last)
1735 return nil
1736 }
1737
1738 func workflowFile(body string) map[string]string {
1739 return map[string]string{".vole/workflows/ci.yaml": body}
1740 }
1741
1742 // TestPushRunsWorkflow is the end-to-end criterion: a push queues a job, a runner
1743 // leases it, executes the pinned steps and reports the log.
1744 func TestPushRunsWorkflow(t *testing.T) {
1745 f := newFixtureWith(t, func(c *config.Config) { c.LeaseSeconds = 2 })
1746 f.publish(f.repo)
1747
1748 work := workRepo(t, workflowFile(`version: 1
1749 jobs:
1750 test:
1751 steps:
1752 - name: say hello
1753 run: echo "hello from the runner"
1754 `))
1755 if _, errs, err := f.git(work, "push", f.sshURL(f.repo), "main:refs/heads/main"); err != nil {
1756 t.Fatalf("push: %s", errs)
1757 }
1758
1759 queued := f.waitForRun(t, 5*time.Second, func(r *runs.Run) bool { return r.Status == runs.StatusQueued })
1760 if queued.Job != "test" || len(queued.Steps) != 1 {
1761 t.Fatalf("queued run = %+v", queued)
1762 }
1763 if queued.WorkflowSHA == "" {
1764 t.Fatal("the workflow blob was not pinned; a later edit could change what runs")
1765 }
1766
1767 cmd := exec.Command(binPath, "runner", "--remote", f.sshBase(), "--once", "--dir", t.TempDir())
1768 cmd.Env = f.ciEnv()
1769 if out, err := cmd.CombinedOutput(); err != nil {
1770 t.Fatalf("runner: %v\n%s", err, out)
1771 }
1772
1773 done := f.waitForRun(t, 10*time.Second, func(r *runs.Run) bool { return r.Status != runs.StatusQueued && r.Status != runs.StatusRunning })
1774 if done.Status != runs.StatusSuccess {
1775 t.Fatalf("run status = %s (exit %d), log:\n%s", done.Status, done.ExitCode, done.Log)
1776 }
1777 if !strings.Contains(done.Log, "hello from the runner") {
1778 t.Fatalf("log does not contain the step output:\n%s", done.Log)
1779 }
1780 if done.SHA != mustSHA(t, work) {
1781 t.Fatalf("run recorded sha %s, want the pushed commit", done.SHA)
1782 }
1783 }
1784
1785 // TestKilledRunnerIsRequeued is the other half of the contract: killing a runner
1786 // mid-job returns the job to the queue and another runner finishes it.
1787 func TestKilledRunnerIsRequeued(t *testing.T) {
1788 if testing.Short() {
1789 t.Skip("waits for a lease to expire and a janitor tick; slow under -race")
1790 }
1791 f := newFixtureWith(t, func(c *config.Config) { c.LeaseSeconds = 2 })
1792 f.publish(f.repo)
1793
1794 work := workRepo(t, workflowFile(`version: 1
1795 jobs:
1796 slow:
1797 steps:
1798 - name: hold the lease
1799 run: sleep 8
1800 `))
1801 if _, errs, err := f.git(work, "push", f.sshURL(f.repo), "main:refs/heads/main"); err != nil {
1802 t.Fatalf("push: %s", errs)
1803 }
1804
1805 first := exec.Command(binPath, "runner", "--remote", f.sshBase(), "--dir", t.TempDir())
1806 first.Env = f.ciEnv()
1807 first.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
1808 first.Stderr = io.Discard
1809 if err := first.Start(); err != nil {
1810 t.Fatal(err)
1811 }
1812 // It must actually own the job before we kill it, or the test proves nothing.
1813 f.waitForRun(t, 20*time.Second, func(r *runs.Run) bool { return r.Status == runs.StatusRunning })
1814 _ = syscall.Kill(-first.Process.Pid, syscall.SIGKILL)
1815 _, _ = first.Process.Wait()
1816
1817 // The lease expires and the janitor puts the job back.
1818 f.waitForRun(t, 30*time.Second, func(r *runs.Run) bool { return r.Status == runs.StatusQueued })
1819
1820 second := exec.Command(binPath, "runner", "--remote", f.sshBase(), "--once", "--dir", t.TempDir())
1821 second.Env = f.ciEnv()
1822 second.Stderr = io.Discard
1823 if err := second.Start(); err != nil {
1824 t.Fatal(err)
1825 }
1826 defer func() { _ = second.Process.Kill(); _, _ = second.Process.Wait() }()
1827
1828 done := f.waitForRun(t, 60*time.Second, func(r *runs.Run) bool {
1829 return r.Status == runs.StatusSuccess || r.Status == runs.StatusFailed
1830 })
1831 if done.Status != runs.StatusSuccess {
1832 t.Fatalf("the retried run did not succeed: %s (exit %d)\n%s", done.Status, done.ExitCode, done.Log)
1833 }
1834 if done.Attempts < 2 {
1835 t.Fatalf("attempts = %d; the job was not actually handed to a second runner", done.Attempts)
1836 }
1837 }
1838
1839 // TestQueuedRunUsesTheWorkflowItWasQueuedFrom: editing a workflow after a push
1840 // must not change what the already-queued run executes. The step list is pinned
1841 // when the run is created, from the blob that was pushed.
1842 func TestQueuedRunUsesTheWorkflowItWasQueuedFrom(t *testing.T) {
1843 f := newFixtureWith(t, func(c *config.Config) { c.LeaseSeconds = 5 })
1844 f.publish(f.repo)
1845
1846 work := workRepo(t, workflowFile(`version: 1
1847 jobs:
1848 test:
1849 steps:
1850 - run: echo first-version
1851 `))
1852 if _, errs, err := f.git(work, "push", f.sshURL(f.repo), "main:refs/heads/main"); err != nil {
1853 t.Fatalf("push v1: %s", errs)
1854 }
1855 firstSHA := mustSHA(t, work)
1856 f.waitForRun(t, 5*time.Second, func(r *runs.Run) bool { return r.SHA == firstSHA })
1857
1858 // Second push rewrites the workflow.
1859 if err := os.WriteFile(filepath.Join(work, ".vole/workflows/ci.yaml"),
1860 []byte("version: 1\njobs:\n test:\n steps:\n - run: echo second-version\n"), 0o644); err != nil {
1861 t.Fatal(err)
1862 }
1863 if _, errs, err := runGit(t, work, "add", "-A"); err != nil {
1864 t.Fatalf("add: %s", errs)
1865 }
1866 if _, errs, err := runGit(t, work, "commit", "-qm", "ci v2"); err != nil {
1867 t.Fatalf("commit: %s", errs)
1868 }
1869 if _, errs, err := f.git(work, "push", f.sshURL(f.repo), "main:refs/heads/main"); err != nil {
1870 t.Fatalf("push v2: %s", errs)
1871 }
1872 secondSHA := mustSHA(t, work)
1873 f.waitForRun(t, 5*time.Second, func(r *runs.Run) bool { return r.SHA == secondSHA })
1874
1875 // Run both jobs.
1876 for i := 0; i < 2; i++ {
1877 cmd := exec.Command(binPath, "runner", "--remote", f.sshBase(), "--once", "--dir", t.TempDir())
1878 cmd.Env = f.ciEnv()
1879 cmd.Stderr = io.Discard
1880 if err := cmd.Start(); err != nil {
1881 t.Fatal(err)
1882 }
1883 _ = cmd.Wait()
1884 }
1885
1886 first := f.waitForRun(t, 30*time.Second, func(r *runs.Run) bool {
1887 return r.SHA == firstSHA && r.Status != runs.StatusQueued && r.Status != runs.StatusRunning
1888 })
1889 second := f.waitForRun(t, 30*time.Second, func(r *runs.Run) bool {
1890 return r.SHA == secondSHA && r.Status != runs.StatusQueued && r.Status != runs.StatusRunning
1891 })
1892 if !strings.Contains(first.Log, "first-version") || strings.Contains(first.Log, "second-version") {
1893 t.Fatalf("the run queued from v1 executed something else:\n%s", first.Log)
1894 }
1895 if !strings.Contains(second.Log, "second-version") {
1896 t.Fatalf("the run queued from v2 did not use its own workflow:\n%s", second.Log)
1897 }
1898 }
1899
1900 func mustSHA(t *testing.T, dir string) string {
1901 t.Helper()
1902 out, _, err := runGit(t, dir, "rev-parse", "HEAD")
1903 if err != nil {
1904 t.Fatal(err)
1905 }
1906 return strings.TrimSpace(out)
1907 }
1908
1696 1909 func mustRepo(t *testing.T, f *fixture) *repo.Meta {
1697 1910 t.Helper()
1698 1911 m, err := f.srv.Repos.Open(f.repo)