package api
import (
"errors"
"sync"
"sync/atomic"
"testing"
"time"
"craftdesign.group/quarry/internal/lease"
"craftdesign.group/quarry/internal/ops"
"craftdesign.group/quarry/internal/zitcli"
)
// The close tag is written and only the ingest after it fails. The
// tag exists, so the close succeeded; and fossil's tag log is
// append-only, so a caller who asks again must not mint a second one.
func TestACloseWhoseIngestFailedIsWrittenOnceAcrossTwoCalls(t *testing.T) {
dir, _, ix, _ := acceptStore(t)
tip := ztip(t, dir, "feat")
writes := 0
r := ops.NewRegistry()
if err := RegisterAll(r, Deps{Version: "v", Repos: NewRepos(map[string]*Repo{"test": {
Index: ix, Zit: zitcli.New(dir),
Guard: func(fn func(lease.Token) error) error { writes++; return fn(testToken) },
Sync: func() (int, error) { return 0, errors.New("ingest round failed") },
}})}); err != nil {
t.Fatal(err)
}
for call := 1; call <= 2; call++ {
out, err := r.Invoke(as("dan", "i"), "checkin.close", checkinCloseIn{Repo: "test", Hash: tip})
if err != nil {
t.Fatalf("FAIL: close call %d = %v; the tag was written, so it succeeded", call, err)
}
if got := out.(checkinCloseOut); !got.Closed || got.Hash != tip {
t.Fatalf("FAIL: close call %d answered %+v", call, got)
}
}
if writes != 1 {
t.Fatalf("FAIL: %d closed tags written across two calls, want 1", writes)
}
t.Log("ok: two calls, one closed tag")
}
// Two closes of one check-in at once: the lease orders nodes, not
// writers on one node, so both could read "not closed" and both write.
// They take turns, and the second finds the first's tag.
func TestTwoSimultaneousClosesWriteOneTag(t *testing.T) {
dir, _, ix, _ := acceptStore(t)
tip := ztip(t, dir, "feat")
var writes atomic.Int32
entered := make(chan struct{})
release := make(chan struct{})
var once sync.Once
r := ops.NewRegistry()
if err := RegisterAll(r, Deps{Version: "v", Repos: NewRepos(map[string]*Repo{"test": {
Index: ix, Zit: zitcli.New(dir),
Guard: func(fn func(lease.Token) error) error {
writes.Add(1)
first := false
once.Do(func() { first = true })
if first {
// Hold the first close between its read and its write.
close(entered)
<-release
}
return fn(testToken)
},
}})}); err != nil {
t.Fatal(err)
}
closeIt := func() error {
_, err := r.Invoke(as("dan", "i"), "checkin.close", checkinCloseIn{Repo: "test", Hash: tip})
return err
}
firstErr := make(chan error, 1)
go func() { firstErr <- closeIt() }()
select {
case <-entered:
case <-time.After(30 * time.Second):
t.Fatal("the first close never reached its write")
}
secondErr := make(chan error, 1)
go func() { secondErr <- closeIt() }()
// An unserialised second close reads, writes and returns while the
// first is held; a serialised one waits here.
select {
case err := <-secondErr:
secondErr <- err
case <-time.After(2 * time.Second):
}
close(release)
if err := <-firstErr; err != nil {
t.Fatalf("FAIL: the first close failed: %v", err)
}
if err := <-secondErr; err != nil {
t.Fatalf("FAIL: the second close failed: %v", err)
}
if n := writes.Load(); n != 1 {
t.Fatalf("FAIL: %d closed tags written by two simultaneous closes, want 1", n)
}
t.Log("ok: two simultaneous closes, one closed tag")
}