internal/control/buildfollow.go

  1package control
  2
  3import (
  4	"errors"
  5	"fmt"
  6	"sync"
  7	"time"
  8
  9	"gitbay.org/gitbay/internal/policy"
 10	"gitbay.org/gitbay/internal/protocol"
 11	"gitbay.org/gitbay/internal/store"
 12)
 13
 14// maxFollows is how many build log follows one account holds open at
 15// once. Signed-out web viewers are account 0 and share it.
 16const maxFollows = 8
 17
 18var (
 19	// followPoll bounds a wait with no wake. A write from another process
 20	// (gitbayd admin, or any session under gitbayd shell) wakes nobody;
 21	// this is how its bytes still arrive.
 22	followPoll = 2 * time.Second
 23	// followSettle is how long a follow keeps reading after the build has
 24	// an outcome: a cancel appends its line after the status changes, and
 25	// a cancelled runner's stream runs on until its next check.
 26	followSettle = time.Second
 27	// followQueued bounds how long a follow waits on a build that stays
 28	// pending: nothing reaps a queued build (ReapStaleBuilds only reaps
 29	// running builds), and a running one is already bounded by the
 30	// reaper's deadline, so a follow needs its own limit for the queued
 31	// case or it never ends.
 32	followQueued = 10 * time.Minute
 33	// followCoalesce is the pause between a wake and the read it causes.
 34	// A runner appends a chunk per read of its output, often a line, and
 35	// each read loads the whole stored log; one read after a short pause
 36	// takes a burst of appends together.
 37	followCoalesce = 200 * time.Millisecond
 38)
 39
 40var (
 41	followMu sync.Mutex
 42	follows  = map[int64]int{}
 43)
 44
 45func takeFollow(uid int64) bool {
 46	followMu.Lock()
 47	defer followMu.Unlock()
 48	if follows[uid] >= maxFollows {
 49		return false
 50	}
 51	follows[uid]++
 52	return true
 53}
 54
 55func dropFollow(uid int64) {
 56	followMu.Lock()
 57	defer followMu.Unlock()
 58	if follows[uid]--; follows[uid] <= 0 {
 59		delete(follows, uid)
 60	}
 61}
 62
 63// mayStillRead reports whether the follower can still read the build's
 64// repository. It looks the repository up by id, so a rename mid-follow
 65// does not end the follow, and reloads the account, so disabling it
 66// does.
 67func mayStillRead(c *Ctx, repoID int64) (bool, error) {
 68	repo, err := c.Store.RepoByID(repoID)
 69	if errors.Is(err, store.ErrNotFound) {
 70		return false, nil
 71	}
 72	if err != nil {
 73		return false, err
 74	}
 75	u := c.User
 76	if u.ID != 0 {
 77		u, err = c.Store.UserByID(u.ID)
 78		if errors.Is(err, store.ErrNotFound) {
 79			return false, nil
 80		}
 81		if err != nil {
 82			return false, err
 83		}
 84		if u.Disabled {
 85			return false, nil
 86		}
 87	}
 88	grant, err := c.Store.AccessRole(repo.ID, u.ID)
 89	if err != nil {
 90		return false, err
 91	}
 92	return policy.CanRead(u, repo, grant), nil
 93}
 94
 95// ended is what a follow returns when its Done closes. A restart says
 96// so, whatever the surface, so the reader knows to follow again; a
 97// reader who left hears nothing.
 98func ended(c *Ctx) int {
 99	select {
100	case <-c.Stopping:
101		fmt.Fprintln(c.Stderr, "gitbay is restarting; follow the build again in a moment")
102	default:
103	}
104	return protocol.ExitFailure
105}
106
107// followBuildLog writes the build's log as it grows and returns once the
108// build has an outcome and its last bytes are written. The outcome goes
109// to stderr, so stdout is the log byte for byte. Read access is checked
110// again every followPoll: a repository made private, or a grant revoked,
111// ends the follow with the answer a new request would get.
112func followBuildLog(c *Ctx, repo store.Repo, b store.Build) int {
113	if !takeFollow(c.User.ID) {
114		return c.fail(protocol.ExitDenied, "%d follows are already open for this account; close one and retry", maxFollows)
115	}
116	defer dropFollow(c.User.ID)
117
118	var off int64
119	var settleBy time.Time
120	var queuedSince time.Time
121	checked := time.Now()
122	for {
123		if time.Since(checked) >= followPoll {
124			ok, err := mayStillRead(c, b.RepoID)
125			if err != nil {
126				return c.fail(protocol.ExitFailure, "%v", err)
127			}
128			if !ok {
129				return c.fail(protocol.ExitNotFound, "repository %s not found", repo.Path())
130			}
131			checked = time.Now()
132		}
133		wake := c.Store.BuildLogWait(b.ID)
134		status, chunk, err := c.Store.BuildLogFrom(b.ID, off)
135		if err != nil {
136			return c.fail(protocol.ExitFailure, "%v", err)
137		}
138		if len(chunk) > 0 {
139			if _, err := c.Stdout.Write(chunk); err != nil {
140				return protocol.ExitFailure
141			}
142			off += int64(len(chunk))
143		}
144		if status == "pending" {
145			if queuedSince.IsZero() {
146				queuedSince = time.Now()
147			}
148		} else {
149			queuedSince = time.Time{}
150		}
151		wait := followPoll
152		if status == "pending" {
153			left := queuedSince.Add(followQueued).Sub(time.Now())
154			if left <= 0 {
155				fmt.Fprintf(c.Stderr, "build %d is still queued; nothing claimed it in %v. Follow again once a runner has.\n", b.Number, followQueued)
156				return protocol.ExitFailure
157			}
158			wait = min(wait, left)
159		}
160		if status != "pending" && status != "running" {
161			if settleBy.IsZero() {
162				settleBy = time.Now().Add(followSettle)
163			}
164			left := time.Until(settleBy)
165			if left <= 0 && len(chunk) == 0 {
166				fmt.Fprintf(c.Stderr, "build %d %s\n", b.Number, status)
167				return protocol.ExitOK
168			}
169			wait = min(wait, max(left, 0))
170		}
171		t := time.NewTimer(wait)
172		select {
173		case <-wake:
174			t.Reset(followCoalesce)
175			select {
176			case <-t.C:
177			case <-c.Done:
178				t.Stop()
179				return ended(c)
180			}
181		case <-t.C:
182		case <-c.Done:
183			t.Stop()
184			return ended(c)
185		}
186		t.Stop()
187	}
188}