lock.go
1package engine
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "os"
8 "path/filepath"
9 "slices"
10 "strings"
11 "syscall"
12 "time"
13
14 "git.kilimanjaro.io/ci/internal/lex"
15)
16
17// Concurrency groups are flock(2) locks on <LocksDir>/<group>.lock. A run
18// takes a group's lock when it first runs an action in that group and holds
19// it until it ends or pauses at a gate (a run waiting for a person must not
20// block others), taking it again when it next runs an action in the group.
21
22func (e *Engine) locksDir() string {
23 if e.Opts.LocksDir != "" {
24 return e.Opts.LocksDir
25 }
26 return filepath.Join(filepath.Dir(e.Dir.Path()), ".locks")
27}
28
29// Locker takes concurrency groups. TryLock doesn't wait: it returns a release
30// function when it got the group, else who holds it.
31type Locker interface {
32 TryLock(ctx context.Context, group, holder string) (release func(), heldBy string, err error)
33}
34
35// FileLocker is flock(2) on <Dir>/<group>.lock, holder written inside.
36type FileLocker struct{ Dir string }
37
38func (l FileLocker) TryLock(_ context.Context, group, holder string) (func(), string, error) {
39 if err := os.MkdirAll(l.Dir, 0o755); err != nil {
40 return nil, "", err
41 }
42 f, err := os.OpenFile(filepath.Join(l.Dir, group+".lock"), os.O_CREATE|os.O_RDWR, 0o644)
43 if err != nil {
44 return nil, "", err
45 }
46 if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil {
47 by, _ := os.ReadFile(f.Name())
48 f.Close()
49 if errors.Is(err, syscall.EWOULDBLOCK) {
50 return nil, strings.TrimSpace(string(by)), nil
51 }
52 return nil, "", fmt.Errorf("lock %s: %w", group, err)
53 }
54 if err := f.Truncate(0); err == nil {
55 _, _ = f.WriteAt([]byte(holder+"\n"), 0)
56 }
57 return func() {
58 _ = syscall.Flock(int(f.Fd()), syscall.LOCK_UN)
59 f.Close()
60 }, "", nil
61}
62
63func (e *Engine) locker() Locker {
64 if e.Opts.Locker != nil {
65 return e.Opts.Locker
66 }
67 return FileLocker{Dir: e.locksDir()}
68}
69
70// acquire blocks until this run holds group, or ctx ends.
71func (e *Engine) acquire(ctx context.Context, group string) error {
72 if _, held := e.locks[group]; held {
73 return nil
74 }
75 told := ""
76 for {
77 release, by, err := e.locker().TryLock(ctx, group, e.st.Meta.Run)
78 if err != nil {
79 return err
80 }
81 if release != nil {
82 if e.locks == nil {
83 e.locks = map[string]func(){}
84 }
85 e.locks[group] = release
86 return nil
87 }
88 if told != by {
89 e.emit(lex.RunEvent{Kind: lex.KindLog, Stream: "stderr",
90 Content: fmt.Sprintf("ci: waiting for concurrency group %s (held by run %s)", group, by)})
91 told = by
92 }
93 select {
94 case <-ctx.Done():
95 return ctx.Err()
96 case <-time.After(time.Second):
97 }
98 }
99}
100
101func (e *Engine) releaseLocks() {
102 for g, release := range e.locks {
103 release()
104 delete(e.locks, g)
105 }
106}
107
108// supersedeGroups returns the concurrency groups of g's options that ask for
109// superseding.
110func (e *Engine) supersedeGroups(g *PendingGate) []string {
111 var out []string
112 for _, o := range g.Options {
113 if c := e.M.Actions[o.Action].Concurrency; c != nil && c.Supersede && !slices.Contains(out, c.Group) {
114 out = append(out, c.Group)
115 }
116 }
117 return out
118}
119
120// supersedeOlder cancels older runs of the same repo and workflow that are
121// waiting at a gate in one of g's superseding concurrency groups: only the
122// newest deploy is worth approving.
123func (e *Engine) supersedeOlder(g *PendingGate) {
124 groups := e.supersedeGroups(g)
125 if len(groups) == 0 {
126 return
127 }
128 runs := filepath.Dir(e.Dir.Path())
129 ents, err := os.ReadDir(runs)
130 if err != nil {
131 return
132 }
133 for _, ent := range ents {
134 d := Dir(filepath.Join(runs, ent.Name()))
135 if d == e.Dir || !ent.IsDir() {
136 continue
137 }
138 st, err := Load(d)
139 if err != nil || st.Status != StatusPaused || st.Gate == nil ||
140 st.Meta.Repo != e.st.Meta.Repo || st.Meta.Workflow != e.st.Meta.Workflow || st.Created >= e.st.Created {
141 continue
142 }
143 other, err := Open(d, nil, Options{})
144 if err != nil {
145 continue
146 }
147 if shared := other.supersedeGroups(st.Gate); slices.ContainsFunc(shared, func(s string) bool { return slices.Contains(groups, s) }) {
148 other.emit(lex.RunEvent{Kind: lex.KindGate, Gate: other.gateRecord(st.Gate, lex.GateSuperseded, Choice{Actor: "run:" + e.st.Meta.Run})})
149 _, _ = other.finish(End{StatusCancelled, lex.ReasonSuper, "superseded by run " + e.st.Meta.Run}, "")
150 e.emit(lex.RunEvent{Kind: lex.KindLog, Stream: "stderr", Content: "ci: superseded run " + st.Meta.Run})
151 }
152 other.Log.Close()
153 }
154}
