package main
import (
"bytes"
"errors"
"log"
"os"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"craftdesign.group/quarry/internal/api"
"craftdesign.group/quarry/internal/oltp"
)
// steppedClock is a registrar clock the test moves by hand.
type steppedClock struct {
mu sync.Mutex
t time.Time
}
func (c *steppedClock) now() time.Time {
c.mu.Lock()
defer c.mu.Unlock()
return c.t
}
func (c *steppedClock) add(d time.Duration) {
c.mu.Lock()
c.t = c.t.Add(d)
c.mu.Unlock()
}
func newClockedRegistrar(t *testing.T) (*registrar, *steppedClock) {
t.Helper()
rg := newTestRegistrar(t, 25)
clk := &steppedClock{t: time.Unix(1_800_000_000, 0)}
rg.clock = clk.now
return rg, clk
}
// eventually polls cond until it holds or two seconds pass.
func eventually(t *testing.T, what string, cond func() bool) {
t.Helper()
deadline := time.Now().Add(2 * time.Second)
for !cond() {
if time.Now().After(deadline) {
t.Fatalf("timed out waiting for %s", what)
}
time.Sleep(time.Millisecond)
}
}
// Another node shares the OLTP and writes it directly. Its change is
// not seen while the cached list is younger than privateTTL, and is
// seen once a read finds the list older than that — no local write.
func TestPrivateListSeesAnotherNodesChangeAfterTheTTL(t *testing.T) {
rg, clk := newClockedRegistrar(t)
if err := rg.odb.AddRepo("vault", t.TempDir(), "", oltp.Public); err != nil {
t.Fatal(err)
}
rg.refreshPrivate()
if rg.IsPrivate("vault") {
t.Fatal("public repository answered private")
}
if err := rg.odb.SetRepoVisibility("vault", oltp.Private); err != nil {
t.Fatal(err)
}
clk.add(privateTTL - time.Millisecond)
for range 3 {
if rg.IsPrivate("vault") {
t.Fatal("a list younger than the TTL was reloaded")
}
}
clk.add(2 * time.Millisecond)
eventually(t, "the other node's change to be seen", func() bool {
return rg.IsPrivate("vault") && !rg.CanRead(t.Context(), "vault")
})
}
// Readers during a slow refresh are served the previous list without
// waiting for it, and however many of them find the list stale, one
// refresh runs.
func TestPrivateListSlowRefreshIsSingleFlightAndNonBlocking(t *testing.T) {
rg, clk := newClockedRegistrar(t)
rg.privLoad = func() (map[string]bool, error) { return map[string]bool{"old": true}, nil }
rg.refreshPrivate()
var loads atomic.Int32
release := make(chan struct{})
rg.privLoad = func() (map[string]bool, error) {
loads.Add(1)
<-release
return map[string]bool{"new": true}, nil
}
clk.add(privateTTL + time.Second)
const readers = 64
var wg sync.WaitGroup
var sawOld atomic.Int32
for range readers {
wg.Add(1)
go func() {
defer wg.Done()
if rg.IsPrivate("old") && !rg.IsPrivate("new") {
sawOld.Add(1)
}
}()
}
done := make(chan struct{})
go func() { wg.Wait(); close(done) }()
select {
case <-done:
case <-time.After(2 * time.Second):
close(release)
t.Fatal("readers blocked on an in-flight refresh")
}
if got := sawOld.Load(); got != readers {
t.Errorf("%d of %d readers saw the previous list during the refresh", got, readers)
}
close(release)
eventually(t, "the refreshed list", func() bool { return rg.IsPrivate("new") })
if got := loads.Load(); got != 1 {
t.Errorf("refreshes run = %d, want 1", got)
}
}
// A refresh that fails leaves the previous list in place, logs once,
// and is not retried by every following read.
func TestPrivateListFailedRefreshKeepsTheOldListAndLogsOnce(t *testing.T) {
var lg syncBuffer
log.SetOutput(&lg)
t.Cleanup(func() { log.SetOutput(os.Stderr) })
rg, clk := newClockedRegistrar(t)
rg.privLoad = func() (map[string]bool, error) { return map[string]bool{"vault": true}, nil }
rg.refreshPrivate()
var loads atomic.Int32
rg.privLoad = func() (map[string]bool, error) {
loads.Add(1)
return nil, errors.New("database is locked")
}
clk.add(privateTTL + time.Second)
var wg sync.WaitGroup
for range 32 {
wg.Add(1)
go func() {
defer wg.Done()
if !rg.IsPrivate("vault") {
t.Error("a failed or in-flight refresh emptied the list")
}
}()
}
wg.Wait()
eventually(t, "the failure to be logged", func() bool {
return strings.Count(lg.String(), "visibility: reload") >= 1
})
for range 8 {
if !rg.IsPrivate("vault") {
t.Fatal("the failed refresh replaced the previous list")
}
}
if got := loads.Load(); got != 1 {
t.Errorf("refresh attempts = %d, want 1", got)
}
if got := strings.Count(lg.String(), "visibility: reload"); got != 1 {
t.Errorf("failure logged %d times, want 1", got)
}
}
// A visibility change made on this node is visible on the very next
// read, and a background refresh that began before it and finishes
// after it cannot put the older answer back.
func TestPrivateListLocalSetIsVisibleImmediately(t *testing.T) {
rg, clk := newClockedRegistrar(t)
if err := rg.odb.AddAccount("dan", "pw12345678", false); err != nil {
t.Fatal(err)
}
if err := rg.odb.AddRepo("vault", t.TempDir(), "", oltp.Public); err != nil {
t.Fatal(err)
}
rg.refreshPrivate()
started := make(chan struct{})
release := make(chan struct{})
var calls atomic.Int32
rg.privLoad = func() (map[string]bool, error) {
set, err := rg.odb.PrivateRepos()
if calls.Add(1) == 1 { // only the background reload stalls
close(started)
<-release
}
return set, err
}
clk.add(privateTTL + time.Second)
if rg.IsPrivate("vault") {
t.Fatal("public repository answered private")
}
select { // a background refresh holds the pre-change list
case <-started:
case <-time.After(2 * time.Second):
t.Fatal("a stale read started no background refresh")
}
private := oltp.Private
if err := rg.Update("vault", api.RepoEdit{Visibility: &private}, "dan"); err != nil {
t.Fatal(err)
}
if !rg.IsPrivate("vault") {
t.Fatal("a local visibility change was not visible on the next read")
}
close(release)
eventually(t, "the background refresh to finish", func() bool { return !rg.privBusy.Load() })
if !rg.IsPrivate("vault") {
t.Fatal("a stale background refresh overwrote the local change")
}
}
// Until a load has succeeded there is no previous list to serve, so
// every repository answers private: a first load that fails at boot
// hides the forge rather than opening its private repositories. The
// TTL retry then recovers the real answer.
func TestPrivateListFailsClosedUntilTheFirstLoadSucceeds(t *testing.T) {
rg, clk := newClockedRegistrar(t)
if err := rg.odb.AddRepo("vault", t.TempDir(), "", oltp.Private); err != nil {
t.Fatal(err)
}
if err := rg.odb.AddRepo("open", t.TempDir(), "", oltp.Public); err != nil {
t.Fatal(err)
}
var calls atomic.Int32
rg.privLoad = func() (map[string]bool, error) {
if calls.Add(1) == 1 {
return nil, errors.New("database is locked")
}
return rg.odb.PrivateRepos()
}
rg.refreshPrivate() // the boot load, failing
if !rg.IsPrivate("vault") {
t.Error("a private repository answered public after a failed first load")
}
if rg.CanRead(t.Context(), "vault") {
t.Error("a private repository was readable anonymously after a failed first load")
}
if !rg.IsPrivate("open") {
t.Error("before any load succeeded, a repository answered public")
}
clk.add(privateTTL + time.Second)
eventually(t, "the retried load to serve the real answer", func() bool {
return !rg.IsPrivate("open") && rg.IsPrivate("vault")
})
if !rg.CanRead(t.Context(), "open") {
t.Error("a public repository is unreadable after a successful load")
}
}
// A load stores its list only if no load that began after it has
// stored one: a slow background reload that finishes last must not
// replace the list a later reload put in place.
func TestPrivateListOlderLoadCannotReplaceANewerOne(t *testing.T) {
rg, clk := newClockedRegistrar(t)
rg.privLoad = func() (map[string]bool, error) { return map[string]bool{"first": true}, nil }
rg.refreshPrivate()
started := make(chan struct{})
release := make(chan struct{})
var calls atomic.Int32
rg.privLoad = func() (map[string]bool, error) {
if calls.Add(1) == 1 {
close(started)
<-release
return map[string]bool{"older": true}, nil
}
return map[string]bool{"newer": true}, nil
}
clk.add(privateTTL + time.Second)
rg.IsPrivate("first") // begins the background load
select {
case <-started:
case <-time.After(2 * time.Second):
t.Fatal("a stale read started no background load")
}
rg.refreshPrivate() // begins after it, stores first
close(release)
eventually(t, "the background load to finish", func() bool { return !rg.privBusy.Load() })
if !rg.IsPrivate("newer") || rg.IsPrivate("older") {
t.Error("an older load replaced the list a newer load stored")
}
}
type syncBuffer struct {
mu sync.Mutex
b bytes.Buffer
}
func (s *syncBuffer) Write(p []byte) (int, error) {
s.mu.Lock()
defer s.mu.Unlock()
return s.b.Write(p)
}
func (s *syncBuffer) String() string {
s.mu.Lock()
defer s.mu.Unlock()
return s.b.String()
}