diff --git a/cmd/objgitd/http.go b/cmd/objgitd/http.go
index a655480..405ff8b 100644
--- a/cmd/objgitd/http.go
+++ b/cmd/objgitd/http.go
@@ -160,7 +160,7 @@ func (d *daemon) handleRPC(w http.ResponseWriter, r *http.Request, service strin
var err error
switch service {
case transport.UploadPackService:
- err = transport.UploadPack(r.Context(), st, in, out, &transport.UploadPackRequest{
+ err = uploadPackHTTP(r.Context(), st, in, out, &transport.UploadPackRequest{
StatelessRPC: true,
GitProtocol: gitProtocol,
})
diff --git a/cmd/objgitd/publication_test.go b/cmd/objgitd/publication_test.go
new file mode 100644
index 0000000..a587b82
--- /dev/null
+++ b/cmd/objgitd/publication_test.go
@@ -0,0 +1,153 @@
+package main
+
+import (
+ "bytes"
+ "context"
+ "errors"
+ "io"
+ "log/slog"
+ "os/exec"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/go-git/go-git/v6/plumbing"
+ "github.com/go-git/go-git/v6/plumbing/protocol/capability"
+ "github.com/go-git/go-git/v6/plumbing/protocol/packp"
+ "github.com/go-git/go-git/v6/plumbing/transport"
+ "github.com/go-git/go-git/v6/storage"
+ "github.com/go-git/go-git/v6/storage/memory"
+ "github.com/go-git/go-git/v6/utils/ioutil"
+)
+
+// gatedRefStorer models the ref-publication contract. The Tigris tests exercise
+// the actual bin/cue PUT barriers; this test checks receive-pack's callback.
+type gatedRefStorer struct {
+ storage.Storer
+ entered chan struct{}
+ release chan struct{}
+ err error
+}
+
+func (s *gatedRefStorer) UpdateReferences(sets []*plumbing.Reference, removes []plumbing.ReferenceName) error {
+ close(s.entered)
+ <-s.release
+ if s.err != nil {
+ return s.err
+ }
+ for _, ref := range sets {
+ if err := s.Storer.SetReference(ref); err != nil {
+ return err
+ }
+ }
+ for _, name := range removes {
+ if err := s.Storer.RemoveReference(name); err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+func TestReceivePackWaitsForPublication(t *testing.T) {
+ if _, err := exec.LookPath("git"); err != nil {
+ t.Skip("git not installed")
+ }
+ work := seedRepo(t)
+ tip, ok := plumbing.FromHex(strings.TrimSpace(runGit(t, work, "rev-parse", "HEAD")))
+ if !ok {
+ t.Fatal("invalid fixture tip")
+ }
+ pack := runGit(t, work, "pack-objects", "--stdout", "--all")
+ for _, tt := range []struct {
+ name string
+ fail, deletion bool
+ }{
+ {name: "success"}, {name: "failed upload", fail: true}, {name: "deletion", deletion: true},
+ } {
+ t.Run(tt.name, func(t *testing.T) {
+ var logs syncBuffer
+ prev := slog.Default()
+ slog.SetDefault(slog.New(slog.NewTextHandler(&logs, nil)))
+ defer slog.SetDefault(prev)
+ s := &gatedRefStorer{Storer: memory.NewStorage(), entered: make(chan struct{}), release: make(chan struct{})}
+ uploadErr := errors.New("pack upload failed")
+ if tt.fail {
+ s.err = uploadErr
+ }
+ zero := plumbing.ZeroHash
+ cmd := &packp.Command{Name: plumbing.NewBranchReferenceName("main"), Old: zero, New: tip}
+ if tt.deletion {
+ if err := s.Storer.SetReference(plumbing.NewHashReference(cmd.Name, tip)); err != nil {
+ t.Fatal(err)
+ }
+ cmd.Old, cmd.New = tip, zero
+ }
+ req := &packp.UpdateRequests{Commands: []*packp.Command{cmd}}
+ req.Capabilities.Set(capability.ReportStatus)
+ var input, output bytes.Buffer
+ if err := req.Encode(&input); err != nil {
+ t.Fatal(err)
+ }
+ if !tt.deletion {
+ input.WriteString(pack)
+ }
+ called := make(chan struct{}, 1)
+ done := make(chan error, 1)
+ finished := make(chan struct{})
+ var releaseOnce sync.Once
+ release := func() { releaseOnce.Do(func() { close(s.release) }) }
+ go func() {
+ defer close(finished)
+ done <- receivePackStreaming(context.Background(), s, io.NopCloser(&input), ioutil.WriteNopCloser(&output), &transport.ReceivePackRequest{StatelessRPC: true}, nil, func(_ io.Writer, _ []refUpdate, _ time.Time) { called <- struct{}{} })
+ }()
+ // Always release the gate before waiting for the request during cleanup.
+ defer func() {
+ release()
+ select {
+ case <-finished:
+ case <-time.After(10 * time.Second):
+ t.Error("receive-pack did not exit")
+ }
+ }()
+ select {
+ case <-s.entered:
+ case err := <-done:
+ t.Fatalf("receive-pack exited before publication: %v", err)
+ case <-time.After(10 * time.Second):
+ t.Fatal("publication not reached")
+ }
+ select {
+ case <-called:
+ t.Fatal("callback ran before publication")
+ default:
+ }
+ release()
+ select {
+ case err := <-done:
+ if tt.fail {
+ if !errors.Is(err, uploadErr) {
+ t.Errorf("receive-pack error = %v, want upload failure", err)
+ }
+ } else if err != nil {
+ t.Fatal(err)
+ }
+ case <-time.After(10 * time.Second):
+ t.Fatal("receive-pack did not finish")
+ }
+ select {
+ case <-called:
+ if tt.fail {
+ t.Error("callback ran after upload failure")
+ }
+ default:
+ if !tt.fail {
+ t.Error("callback missing after publication")
+ }
+ }
+ if tt.fail && !strings.Contains(output.String(), uploadErr.Error()) {
+ t.Errorf("missing failure status: %q", output.String())
+ }
+ })
+ }
+}
diff --git a/cmd/objgitd/receivepack.go b/cmd/objgitd/receivepack.go
index 39af635..7a9078a 100644
--- a/cmd/objgitd/receivepack.go
+++ b/cmd/objgitd/receivepack.go
@@ -188,6 +188,12 @@ func receivePackStreaming(
return res
}
+ // updateReferences commits every ref through the storer's SetReference
+ // family. The tigris storer flushes every queued pack upload (bin and
+ // cue) before it publishes a ref, so by the time onUpdated runs, and a
+ // hook can clone the ref, the packs are durable. receive-pack adds no
+ // barrier of its own. TestTigrisPushPublication in s3_publication_test.go
+ // holds the fake S3's uploads open and checks that ordering.
var firstErr error
cmdStatus := make(map[plumbing.ReferenceName]error)
acceptedAt := updateReferences(st, updreq, cmdStatus, &firstErr)
diff --git a/cmd/objgitd/s3_publication_test.go b/cmd/objgitd/s3_publication_test.go
new file mode 100644
index 0000000..e98f750
--- /dev/null
+++ b/cmd/objgitd/s3_publication_test.go
@@ -0,0 +1,319 @@
+package main
+
+import (
+ "context"
+ "encoding/xml"
+ "fmt"
+ "io"
+ "log/slog"
+ "net"
+ "net/http"
+ "net/http/httptest"
+ "os/exec"
+ "path/filepath"
+ "sort"
+ "strconv"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/aws/aws-sdk-go-v2/aws"
+ "github.com/aws/aws-sdk-go-v2/credentials"
+ "github.com/aws/aws-sdk-go-v2/service/s3"
+ "github.com/go-git/go-billy/v6/memfs"
+ "github.com/tigrisdata/objgit/internal/auth"
+ "github.com/tigrisdata/objgit/internal/repofs"
+ "github.com/tigrisdata/objgit/internal/storage/tigris"
+)
+
+// publicationS3 is an HTTP S3 fixture, so these tests exercise the real SDK,
+// Tigris uploader, reference transaction, transports, and hook dispatcher.
+type publicationS3 struct {
+ mu sync.Mutex
+ objects map[string][]byte
+ etags map[string]string
+ seq int
+ binStarted, cueStarted chan struct{}
+ bins, cues chan struct{}
+ binOnce, cueOnce sync.Once
+ fail string
+ // earlyRefs records every ref write that arrived while a bin or cue
+ // upload of this push was still held open. The storer must flush its
+ // upload queue before it publishes a ref, so this list must stay empty.
+ earlyRefs []string
+}
+
+func closed(ch <-chan struct{}) bool {
+ select {
+ case <-ch:
+ return true
+ default:
+ return false
+ }
+}
+
+// earlyRefWrites returns the ref keys written before the uploads were released.
+func (s *publicationS3) earlyRefWrites() []string {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ return append([]string(nil), s.earlyRefs...)
+}
+
+func (s *publicationS3) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+ key := strings.TrimPrefix(r.URL.Path, "/bucket/")
+ if r.Method == http.MethodPut && (strings.HasSuffix(key, ".bin") || strings.HasSuffix(key, ".cue")) {
+ if strings.HasSuffix(key, ".bin") {
+ s.binOnce.Do(func() { close(s.binStarted) })
+ select {
+ case <-s.bins:
+ case <-r.Context().Done():
+ return
+ }
+ } else {
+ s.cueOnce.Do(func() { close(s.cueStarted) })
+ select {
+ case <-s.cues:
+ case <-r.Context().Done():
+ return
+ }
+ }
+ if s.fail != "" && strings.HasSuffix(key, s.fail) {
+ w.WriteHeader(http.StatusForbidden)
+ fmt.Fprint(w, "AccessDeniedinjected upload failure")
+ return
+ }
+ }
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ isRef := strings.HasSuffix(key, "packed-refs") || strings.Contains(key, "refs/")
+ if r.Method == http.MethodPut && isRef && closed(s.binStarted) && (!closed(s.bins) || !closed(s.cues)) {
+ s.earlyRefs = append(s.earlyRefs, key)
+ }
+ if r.Method == http.MethodGet && r.URL.Query().Get("list-type") == "2" {
+ type entry struct {
+ Key string
+ Size int
+ }
+ out := struct {
+ XMLName xml.Name `xml:"ListBucketResult"`
+ IsTruncated bool
+ Contents []entry
+ }{}
+ keys := make([]string, 0, len(s.objects))
+ for k := range s.objects {
+ keys = append(keys, k)
+ }
+ sort.Strings(keys)
+ for _, k := range keys {
+ if strings.HasPrefix(k, r.URL.Query().Get("prefix")) {
+ out.Contents = append(out.Contents, entry{k, len(s.objects[k])})
+ }
+ }
+ w.Header().Set("Content-Type", "application/xml")
+ _ = xml.NewEncoder(w).Encode(out)
+ return
+ }
+ body, exists := s.objects[key]
+ switch r.Method {
+ case http.MethodPut:
+ if (r.Header.Get("If-None-Match") == "*" && exists) || (r.Header.Get("If-Match") != "" && r.Header.Get("If-Match") != s.etags[key]) {
+ w.WriteHeader(412)
+ fmt.Fprint(w, "PreconditionFailed")
+ return
+ }
+ data, err := io.ReadAll(r.Body)
+ if err != nil {
+ http.Error(w, err.Error(), 500)
+ return
+ }
+ s.seq++
+ s.objects[key] = data
+ s.etags[key] = fmt.Sprintf("\"%d\"", s.seq)
+ w.Header().Set("ETag", s.etags[key])
+ case http.MethodGet, http.MethodHead:
+ if !exists {
+ w.WriteHeader(404)
+ fmt.Fprint(w, "NoSuchKey")
+ return
+ }
+ w.Header().Set("ETag", s.etags[key])
+ if rng := r.Header.Get("Range"); rng != "" {
+ var start, end int
+ if _, err := fmt.Sscanf(rng, "bytes=%d-%d", &start, &end); err != nil || start < 0 || end < start || end >= len(body) {
+ w.WriteHeader(416)
+ return
+ }
+ w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, end, len(body)))
+ body = body[start : end+1]
+ w.Header().Set("Content-Length", strconv.Itoa(len(body)))
+ w.WriteHeader(206)
+ } else {
+ w.Header().Set("Content-Length", strconv.Itoa(len(body)))
+ }
+ if r.Method == http.MethodGet {
+ _, _ = w.Write(body)
+ }
+ case http.MethodDelete:
+ delete(s.objects, key)
+ delete(s.etags, key)
+ w.WriteHeader(204)
+ default:
+ http.Error(w, "unexpected S3 operation", 400)
+ }
+}
+
+func TestTigrisPushPublication(t *testing.T) {
+ if _, err := exec.LookPath("git"); err != nil {
+ t.Skip("git not installed")
+ }
+ for _, tt := range []struct{ name, transport, fail string }{
+ {"http success", "http", ""}, {"http bin failure", "http", ".bin"}, {"http cue failure", "http", ".cue"},
+ {"ssh success", "ssh", ""}, {"ssh bin failure", "ssh", ".bin"}, {"ssh cue failure", "ssh", ".cue"},
+ } {
+ t.Run(tt.name, func(t *testing.T) {
+ if tt.transport == "ssh" {
+ checkSSHBinaries(t)
+ }
+ backend := &publicationS3{objects: map[string][]byte{}, etags: map[string]string{}, binStarted: make(chan struct{}), cueStarted: make(chan struct{}), bins: make(chan struct{}), cues: make(chan struct{}), fail: tt.fail}
+ var binRelease, cueRelease sync.Once
+ releaseBins := func() { binRelease.Do(func() { close(backend.bins) }) }
+ releaseCues := func() { cueRelease.Do(func() { close(backend.cues) }) }
+ // Release blocked requests before closing the HTTP fixture, even on failure.
+ endpoint := httptest.NewServer(backend)
+ defer endpoint.Close()
+ defer releaseBins()
+ defer releaseCues()
+ client := s3.New(s3.Options{Region: "auto", BaseEndpoint: aws.String(endpoint.URL), UsePathStyle: true, Credentials: credentials.NewStaticCredentialsProvider("test", "test", ""), RequestChecksumCalculation: aws.RequestChecksumCalculationWhenRequired, RetryMaxAttempts: 1})
+ base, err := tigris.New(context.Background(), "bucket", tigris.WithS3Client(client))
+ if err != nil {
+ t.Fatal(err)
+ }
+ logs := &syncBuffer{}
+ prev := slog.Default()
+ slog.SetDefault(slog.New(slog.NewTextHandler(logs, nil)))
+ defer slog.SetDefault(prev)
+ d := &daemon{sysFS: memfs.New(), resolver: repofs.BucketResolver{Base: tigrisBase{s: base}}, authz: auth.AllowAnonymous{AllowWrite: true}, allowHooks: true, hookTimeout: 10 * time.Second}
+ var remote string
+ var gitEnv []string
+ if tt.transport == "http" {
+ server := httptest.NewServer(d.httpHandler())
+ defer server.Close()
+ remote = server.URL + "/test/publication.git"
+ } else {
+ server, err := newSSHServer(d, "")
+ if err != nil {
+ t.Fatal(err)
+ }
+ ln, err := net.Listen("tcp", "127.0.0.1:0")
+ if err != nil {
+ t.Fatal(err)
+ }
+ go server.Serve(ln)
+ defer server.Close()
+ remote = "ssh://git@" + ln.Addr().String() + "/test/publication.git"
+ gitEnv = gitSSHEnv(t)
+ }
+ defer releaseBins()
+ defer releaseCues()
+ git := func(dir string, args ...string) (string, error) {
+ if tt.transport == "ssh" {
+ return gitWithEnv(dir, gitEnv, args...)
+ }
+ return tryGit(dir, args...)
+ }
+ work := seedRepo(t)
+ writeFile(t, filepath.Join(work, ".objgit", "hooks", "receive-pack"), "echo publication-hook-ran\n")
+ writeFile(t, filepath.Join(work, "payload"), strings.Repeat("pack content\n", 1000))
+ runGit(t, work, "add", ".")
+ runGit(t, work, "commit", "-m", "hook fixture")
+ type result struct {
+ out string
+ err error
+ }
+ done := make(chan result, 1)
+ go func() { out, err := git(work, "push", remote, "main"); done <- result{out, err} }()
+ wait := func(ch <-chan struct{}) {
+ t.Helper()
+ select {
+ case <-ch:
+ case res := <-done:
+ t.Fatalf("push ended before gate: %v\n%s", res.err, res.out)
+ case <-time.After(15 * time.Second):
+ t.Fatal("upload did not reach gate")
+ }
+ }
+ wait(backend.binStarted)
+ assertNoHook := func() {
+ t.Helper()
+ if strings.Contains(logs.String(), "hook: running") {
+ t.Fatal("hook ran before uploads completed")
+ }
+ }
+ assertNoHook()
+ select {
+ case <-backend.cueStarted:
+ t.Fatal("cue started before bin completed")
+ default:
+ }
+ releaseBins()
+ if tt.fail != ".bin" {
+ wait(backend.cueStarted)
+ assertNoHook()
+ }
+ releaseCues()
+ var res result
+ select {
+ case res = <-done:
+ case <-time.After(15 * time.Second):
+ t.Fatal("push did not finish")
+ }
+ // The fake S3 saw every ref write. None may land while a pack
+ // upload is still held open: the storer flushes first.
+ if early := backend.earlyRefWrites(); len(early) > 0 {
+ t.Fatalf("refs written before uploads completed: %v", early)
+ }
+ if tt.fail != "" {
+ if res.err == nil {
+ t.Fatalf("failed upload accepted: %s", res.out)
+ }
+ assertNoHook()
+ out, err := git(work, "ls-remote", remote, "refs/heads/main")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if strings.Contains(out, "\trefs/heads/main") {
+ t.Fatalf("failed upload published main: %s", out)
+ }
+ return
+ }
+ if res.err != nil {
+ t.Fatalf("push: %v\n%s", res.err, res.out)
+ }
+ if !strings.Contains(res.out, "publication-hook-ran") {
+ t.Fatalf("hook output missing: %s", res.out)
+ }
+ clone := filepath.Join(t.TempDir(), "clone")
+ if out, err := git(work, "clone", "--depth=1", "--no-checkout", remote, clone); err != nil {
+ t.Fatalf("cold clone: %v\n%s\nlogs:\n%s", err, out, logs.String())
+ }
+ if got, want := runGit(t, clone, "rev-parse", "HEAD"), runGit(t, work, "rev-parse", "HEAD"); got != want {
+ t.Fatalf("cloned tip=%s want=%s", got, want)
+ }
+ runGit(t, clone, "fsck", "--full")
+ writeFile(t, filepath.Join(work, "second.txt"), "second push\n")
+ runGit(t, work, "add", "second.txt")
+ runGit(t, work, "commit", "-m", "second push")
+ if out, err := git(work, "push", remote, "main"); err != nil {
+ t.Fatalf("second push: %v\n%s", err, out)
+ }
+ if out, err := git(clone, "fetch", "--depth=1", "origin", "main"); err != nil {
+ t.Fatalf("shallow fetch: %v\n%s", err, out)
+ }
+ if got, want := runGit(t, clone, "rev-parse", "FETCH_HEAD"), runGit(t, work, "rev-parse", "HEAD"); got != want {
+ t.Fatalf("fetched tip=%s want=%s", got, want)
+ }
+ runGit(t, clone, "fsck", "--full")
+ })
+ }
+}
diff --git a/cmd/objgitd/uploadpack.go b/cmd/objgitd/uploadpack.go
new file mode 100644
index 0000000..84ddbbe
--- /dev/null
+++ b/cmd/objgitd/uploadpack.go
@@ -0,0 +1,402 @@
+package main
+
+import (
+ "bufio"
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "math"
+
+ "github.com/go-git/go-git/v6/config"
+ "github.com/go-git/go-git/v6/plumbing"
+ "github.com/go-git/go-git/v6/plumbing/format/packfile"
+ "github.com/go-git/go-git/v6/plumbing/format/pktline"
+ "github.com/go-git/go-git/v6/plumbing/object"
+ "github.com/go-git/go-git/v6/plumbing/protocol"
+ "github.com/go-git/go-git/v6/plumbing/protocol/capability"
+ "github.com/go-git/go-git/v6/plumbing/protocol/packp"
+ "github.com/go-git/go-git/v6/plumbing/protocol/packp/sideband"
+ "github.com/go-git/go-git/v6/plumbing/revlist"
+ "github.com/go-git/go-git/v6/plumbing/transport"
+ "github.com/go-git/go-git/v6/storage"
+ "github.com/go-git/go-git/v6/utils/ioutil"
+)
+
+// uploadPackHTTP is a fork of go-git's transport.UploadPack (v6) that adds
+// one seam for Git's shallow discovery round over stateless Smart HTTP.
+//
+// For a shallow clone or fetch, Git first POSTs wants + deepen + flush with no
+// haves and no done, and expects the response to end at the shallow-update
+// flush. Upstream instead goes on to decode UploadHaves, which hits EOF, and
+// with multi_ack it appends a NAK that Git mistakes for the next round's
+// shallow list. The seam sits right after the first-round upload-request
+// decode: in stateless mode, when the request carried a deepen, it peeks the
+// buffered reader for EOF before the reader is closed. If the body is
+// exhausted, the shallow update is written as usual and the call returns
+// without decoding haves and without writing a NAK. A request that carries
+// haves or done after the deepen takes the normal path, and deepen-since /
+// deepen-not still get upstream's "unsupported depth" error.
+//
+// Everything except that seam mirrors transport.UploadPack verbatim,
+// including the helper functions copied below.
+func uploadPackHTTP(
+ ctx context.Context,
+ st storage.Storer,
+ r io.ReadCloser,
+ w io.WriteCloser,
+ opts *transport.UploadPackRequest,
+) error {
+ if w == nil {
+ return fmt.Errorf("nil writer")
+ }
+
+ w = ioutil.NewContextWriteCloser(ctx, w)
+
+ if opts == nil {
+ opts = &transport.UploadPackRequest{}
+ }
+
+ if opts.AdvertiseRefs || !opts.StatelessRPC {
+ switch version := transport.ProtocolVersion(opts.GitProtocol); version {
+ case protocol.V1:
+ if _, err := pktline.Writef(w, "version %d\n", version); err != nil {
+ return err
+ }
+ // TODO: support version 2
+ case protocol.V0, protocol.V2:
+ default:
+ return fmt.Errorf("%w: %q", transport.ErrUnsupportedVersion, version)
+ }
+
+ if err := transport.AdvertiseRefs(ctx, st, w, transport.UploadPackService, opts.StatelessRPC); err != nil {
+ return fmt.Errorf("advertising references: %w", err)
+ }
+ }
+
+ if opts.AdvertiseRefs {
+ // Done, there's nothing else to do
+ return nil
+ }
+
+ if r == nil {
+ return fmt.Errorf("nil reader")
+ }
+
+ r = ioutil.NewContextReadCloser(ctx, r)
+
+ rd := bufio.NewReader(r)
+ l, _, err := pktline.PeekLine(rd)
+ if err != nil {
+ return fmt.Errorf("peeking line: %w", err)
+ }
+
+ // In case the client has nothing to send, it sends a flush packet to
+ // indicate that it is done sending data. In that case, we're done
+ // here.
+ if l == pktline.Flush {
+ return nil
+ }
+
+ var done bool
+ var haves []plumbing.Hash
+ var upreq *packp.UploadRequest
+ var havesWithRef map[plumbing.Hash][]plumbing.Hash
+ var multiAck, multiAckDetailed bool
+ var caps capability.List
+ var wants []plumbing.Hash
+ var ack packp.ACK
+ firstRound := true
+ for !done {
+ writec := make(chan error)
+ if firstRound || opts.StatelessRPC {
+ upreq = &packp.UploadRequest{}
+ if err := upreq.Decode(rd); err != nil {
+ return fmt.Errorf("decoding upload-request: %w", err)
+ }
+
+ wants = upreq.Wants
+ caps = upreq.Capabilities
+
+ // Seam: a stateless shallow request whose body ends at the
+ // upload-request flush is Git's discovery round. Peek before
+ // r.Close below, because an HTTP body may refuse reads after Close.
+ discoveryRound := false
+ if opts.StatelessRPC && !upreq.Depth.IsZero() {
+ if _, err := rd.Peek(1); err != nil {
+ if !errors.Is(err, io.EOF) {
+ return fmt.Errorf("peeking after upload-request: %w", err)
+ }
+ discoveryRound = true
+ }
+ }
+
+ if err := r.Close(); err != nil {
+ return fmt.Errorf("closing reader: %w", err)
+ }
+
+ // Find common commits/objects
+ havesWithRef, err = revlist.ObjectsWithRef(st, wants, nil)
+ if err != nil {
+ return fmt.Errorf("getting objects with ref: %w", err)
+ }
+
+ // Encode objects to packfile and write to client
+ multiAck = caps.Supports(capability.MultiACK)
+ multiAckDetailed = caps.Supports(capability.MultiACKDetailed)
+
+ go func() {
+ // TODO: support deepen-since, and deepen-not
+ var shupd packp.ShallowUpdate
+ if !upreq.Depth.IsZero() {
+ if upreq.Depth.Deepen > 0 {
+ if err := getShallowCommits(st, wants, upreq.Depth.Deepen, &shupd); err != nil {
+ writec <- fmt.Errorf("getting shallow commits: %w", err)
+ return
+ }
+ } else {
+ writec <- fmt.Errorf("unsupported depth: %+v", upreq.Depth)
+ return
+ }
+
+ if err := shupd.Encode(w); err != nil {
+ writec <- fmt.Errorf("sending shallow-update: %w", err)
+ return
+ }
+ }
+
+ writec <- nil
+ }()
+
+ if err := <-writec; err != nil {
+ return err
+ }
+
+ // Seam: the shallow update is the whole response for a discovery
+ // round. Git sends the haves and done in its next request.
+ if discoveryRound {
+ return nil
+ }
+ }
+
+ var uphav packp.UploadHaves
+ if err := uphav.Decode(rd); err != nil {
+ return fmt.Errorf("decoding upload-haves: %w", err)
+ }
+
+ if err := r.Close(); err != nil {
+ return fmt.Errorf("closing reader: %w", err)
+ }
+
+ haves = append(haves, uphav.Haves...)
+ done = uphav.Done
+
+ var acks []packp.ACK
+ for _, hu := range uphav.Haves {
+ _, ok := havesWithRef[hu]
+
+ var status packp.ACKStatus
+ if multiAckDetailed {
+ status = packp.ACKCommon
+ if !ok {
+ status = packp.ACKReady
+ }
+ } else if multiAck {
+ status = packp.ACKContinue
+ }
+
+ if ok || multiAck || multiAckDetailed {
+ ack = packp.ACK{Hash: hu, Status: status}
+ acks = append(acks, ack)
+ if !multiAck && !multiAckDetailed {
+ break
+ }
+ }
+ }
+
+ go func() {
+ defer close(writec)
+
+ if len(haves) > 0 {
+ // Encode ACKs to client when we have haves
+ srvrsp := packp.ServerResponse{ACKs: acks}
+ if err := srvrsp.Encode(w); err != nil {
+ writec <- fmt.Errorf("sending acks server-response: %w", err)
+ return
+ }
+ }
+
+ switch {
+ case !done:
+ if multiAck || multiAckDetailed {
+ // Encode a NAK for multi-ack
+ srvrsp := packp.ServerResponse{}
+ if err := srvrsp.Encode(w); err != nil {
+ writec <- fmt.Errorf("sending nak server-response: %w", err)
+ return
+ }
+ }
+ case !ack.Hash.IsZero() && (multiAck || multiAckDetailed):
+ // We're done, send the final ACK
+ ack.Status = 0
+ srvrsp := packp.ServerResponse{ACKs: []packp.ACK{ack}}
+ if err := srvrsp.Encode(w); err != nil {
+ writec <- fmt.Errorf("sending final ack server-response: %w", err)
+ return
+ }
+ case ack.Hash.IsZero() && len(haves) == 0:
+ // No haves were sent. Emit the single terminal NAK.
+ //
+ // When haves *were* sent, the ServerResponse{ACKs: acks}
+ // write above already emitted a NAK (encodeServerResponse
+ // writes NAK when ACKs is empty). Emitting another one here
+ // would produce two consecutive "0008NAK\n" pktlines;
+ // ServerResponse.Decode consumes only the first, and the
+ // second would then be misread by the sideband demuxer as
+ // a frame with channel byte 'N' ("unknown channel NAK").
+ srvrsp := packp.ServerResponse{}
+ if err := srvrsp.Encode(w); err != nil {
+ writec <- fmt.Errorf("sending final nak server-response: %w", err)
+ return
+ }
+ }
+
+ writec <- nil
+ }()
+
+ if err := <-writec; err != nil {
+ return err
+ }
+
+ firstRound = false
+ }
+
+ // Done with the request, now close the reader
+ // to indicate that we are done reading from it.
+ if err := r.Close(); err != nil {
+ return fmt.Errorf("closing reader: %w", err)
+ }
+
+ objs, err := objectsToUpload(st, wants, haves)
+ if err != nil {
+ _ = w.Close()
+ return fmt.Errorf("getting objects to upload: %w", err)
+ }
+
+ var (
+ useSideband bool
+ writer io.Writer = w
+ )
+ if caps.Supports(capability.Sideband64k) {
+ writer = sideband.NewMuxer(sideband.Sideband64k, w)
+ useSideband = true
+ } else if caps.Supports(capability.Sideband) {
+ writer = sideband.NewMuxer(sideband.Sideband, w)
+ useSideband = true
+ }
+
+ // TODO: Support shallow-file
+ // TODO: Support thin-pack
+ var packWindow uint
+ if opts.SkipDeltaCompression {
+ packWindow = 0
+ } else if cfg, cerr := st.Config(); cerr == nil && cfg != nil {
+ packWindow = cfg.Pack.Window
+ } else {
+ packWindow = config.DefaultPackWindow
+ }
+
+ e := packfile.NewEncoder(writer, st, false)
+ _, err = e.Encode(objs, packWindow)
+ if err != nil {
+ return fmt.Errorf("encoding packfile: %w", err)
+ }
+
+ if useSideband {
+ if err := pktline.WriteFlush(w); err != nil {
+ return fmt.Errorf("flushing sideband: %w", err)
+ }
+ }
+
+ if err := w.Close(); err != nil {
+ return fmt.Errorf("closing writer: %w", err)
+ }
+
+ return nil
+}
+
+// The helpers below are copied verbatim from go-git's transport package
+// (plumbing/transport/upload_pack.go), which keeps them unexported.
+
+func objectsToUpload(st storage.Storer, wants, haves []plumbing.Hash) ([]plumbing.Hash, error) {
+ return revlist.Objects(st, wants, haves)
+}
+
+func getShallowCommits(st storage.Storer, heads []plumbing.Hash, depth int, upd *packp.ShallowUpdate) error {
+ var i, curDepth int
+ var commit *object.Commit
+ depths := map[*object.Commit]int{}
+ stack := []object.Object{}
+
+ for commit != nil || i < len(heads) || len(stack) > 0 {
+ if commit == nil {
+ if i < len(heads) {
+ obj, err := st.EncodedObject(plumbing.CommitObject, heads[i])
+ i++
+ if err != nil {
+ continue
+ }
+
+ commit, err = object.DecodeCommit(st, obj)
+ if err != nil {
+ commit = nil
+ continue
+ }
+
+ depths[commit] = 0
+ curDepth = 0
+ } else if len(stack) > 0 {
+ commit = stack[len(stack)-1].(*object.Commit)
+ stack = stack[:len(stack)-1]
+ curDepth = depths[commit]
+ }
+ }
+
+ curDepth++
+
+ if depth != math.MaxInt && curDepth >= depth {
+ upd.Shallows = append(upd.Shallows, commit.Hash)
+ commit = nil
+ continue
+ }
+
+ upd.Unshallows = append(upd.Unshallows, commit.Hash)
+
+ parents := commit.Parents()
+ commit = nil
+ for {
+ parent, err := parents.Next()
+ if err == io.EOF {
+ break
+ }
+ if err != nil {
+ return err
+ }
+
+ if depths[parent] != 0 && curDepth >= depths[parent] {
+ continue
+ }
+
+ depths[parent] = curDepth
+
+ if _, err := parents.Next(); err == nil {
+ stack = append(stack, parent)
+ } else {
+ commit = parent
+ curDepth = depths[commit]
+ }
+ }
+ }
+
+ return nil
+}
diff --git a/cmd/objgitd/uploadpack_test.go b/cmd/objgitd/uploadpack_test.go
new file mode 100644
index 0000000..0d5f541
--- /dev/null
+++ b/cmd/objgitd/uploadpack_test.go
@@ -0,0 +1,149 @@
+package main
+
+import (
+ "bytes"
+ "context"
+ "fmt"
+ "io"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/go-git/go-git/v6/plumbing"
+ "github.com/go-git/go-git/v6/plumbing/object"
+ "github.com/go-git/go-git/v6/plumbing/transport"
+ "github.com/go-git/go-git/v6/storage/memory"
+ "github.com/go-git/go-git/v6/utils/ioutil"
+)
+
+// pkt frames one pkt-line payload.
+func pkt(payload string) string {
+ return fmt.Sprintf("%04x%s", len(payload)+4, payload)
+}
+
+// seedCommitChain stores n commits on an empty tree, each the parent of the
+// next, and returns them oldest first.
+func seedCommitChain(t *testing.T, st *memory.Storage, n int) []plumbing.Hash {
+ t.Helper()
+ to := st.NewEncodedObject()
+ if err := (&object.Tree{}).Encode(to); err != nil {
+ t.Fatal(err)
+ }
+ th, err := st.SetEncodedObject(to)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var chain []plumbing.Hash
+ for i := range n {
+ sig := object.Signature{Name: "T", Email: "t@example.com", When: time.Unix(int64(i), 0)}
+ c := &object.Commit{Author: sig, Committer: sig, Message: fmt.Sprint(i), TreeHash: th}
+ if i > 0 {
+ c.ParentHashes = []plumbing.Hash{chain[i-1]}
+ }
+ co := st.NewEncodedObject()
+ if err := c.Encode(co); err != nil {
+ t.Fatal(err)
+ }
+ ch, err := st.SetEncodedObject(co)
+ if err != nil {
+ t.Fatal(err)
+ }
+ chain = append(chain, ch)
+ }
+ return chain
+}
+
+func TestUploadPackHTTP(t *testing.T) {
+ st := memory.NewStorage()
+ chain := seedCommitChain(t, st, 3)
+ tip := chain[2].String()
+ shallowTip := pkt("shallow "+tip+"\n") + "0000"
+ const nak = "0008NAK\n"
+
+ for _, tt := range []struct {
+ name string
+ stateless bool
+ input string
+ want string // the whole response, or its prefix when pack is set
+ pack bool // a packfile follows want
+ wantErr string
+ }{
+ {
+ name: "shallow discovery round",
+ stateless: true,
+ input: pkt("want "+tip+" multi_ack_detailed side-band-64k\n") + pkt("deepen 1\n") + "0000",
+ want: shallowTip,
+ },
+ {
+ name: "shallow discovery round without multi_ack",
+ stateless: true,
+ input: pkt("want "+tip+"\n") + pkt("deepen 1\n") + "0000",
+ want: shallowTip,
+ },
+ {
+ name: "shallow with done in one round",
+ stateless: true,
+ input: pkt("want "+tip+"\n") + pkt("deepen 1\n") + "0000" + pkt("done\n"),
+ want: shallowTip + nak + "PACK",
+ pack: true,
+ },
+ {
+ name: "shallow with haves and done in one round",
+ stateless: true,
+ input: pkt("want "+tip+"\n") + pkt("deepen 1\n") + "0000" + pkt("have "+chain[0].String()+"\n") + pkt("done\n"),
+ want: shallowTip + pkt("ACK "+chain[0].String()+"\n") + "PACK",
+ pack: true,
+ },
+ {
+ name: "ordinary request",
+ stateless: true,
+ input: pkt("want "+tip+"\n") + "0000" + pkt("done\n"),
+ want: nak + "PACK",
+ pack: true,
+ },
+ {
+ name: "deepen-since is unsupported",
+ stateless: true,
+ input: pkt("want "+tip+"\n") + pkt("deepen-since 1234567890\n") + "0000",
+ wantErr: "unsupported depth",
+ },
+ {
+ name: "deepen-not is unsupported",
+ stateless: true,
+ input: pkt("want "+tip+"\n") + pkt("deepen-not main\n") + "0000",
+ wantErr: "unsupported depth",
+ },
+ {
+ name: "flush only",
+ stateless: true,
+ input: "0000",
+ },
+ } {
+ t.Run(tt.name, func(t *testing.T) {
+ var out bytes.Buffer
+ err := uploadPackHTTP(context.Background(), st,
+ io.NopCloser(strings.NewReader(tt.input)),
+ ioutil.WriteNopCloser(&out),
+ &transport.UploadPackRequest{StatelessRPC: tt.stateless})
+ if tt.wantErr != "" {
+ if err == nil || !strings.Contains(err.Error(), tt.wantErr) {
+ t.Fatalf("err=%v want %q", err, tt.wantErr)
+ }
+ return
+ }
+ if err != nil {
+ t.Fatalf("err=%v\nresponse=%q", err, out.String())
+ }
+ got := out.String()
+ if tt.pack {
+ if !strings.HasPrefix(got, tt.want) {
+ t.Fatalf("response=%q\nwant prefix %q", got, tt.want)
+ }
+ return
+ }
+ if got != tt.want {
+ t.Fatalf("response=%q\nwant %q", got, tt.want)
+ }
+ })
+ }
+}
diff --git a/docs/architecture/hooks.md b/docs/architecture/hooks.md
index 4e2c4c8..53426f7 100644
--- a/docs/architecture/hooks.md
+++ b/docs/architecture/hooks.md
@@ -7,6 +7,16 @@ Hook output streams to the pushing client live, as `remote: ...` lines. Hooks
therefore run **synchronously**: the client waits for them to finish before
the push completes. `-hook-timeout` bounds that wait.
+The Tigris storer waits for every queued bin and cue upload before it
+publishes a ref. `updateReferences` commits the refs through `SetReference`,
+and the storer flushes its upload queue before it writes a ref. The hook runs
+after that commit, so the packs are durable before a hook can clone the ref.
+If an upload fails, the storer does not publish the ref and the hook does not
+run. The upload failure remains fatal for that writer, and later ref updates
+on the same writer return the same error. `TestTigrisPushPublication` in
+`cmd/objgitd/s3_publication_test.go` holds the uploads open and checks this
+order.
+
A hook **cannot reject a push**. It runs after the refs are updated and after
report-status is sent, which is post-receive semantics. Deleted branches are
skipped.
diff --git a/docs/architecture/tigris-storer.md b/docs/architecture/tigris-storer.md
index 4bc951b..fd63099 100644
--- a/docs/architecture/tigris-storer.md
+++ b/docs/architecture/tigris-storer.md
@@ -285,7 +285,13 @@ failure of one repository can therefore never block or poison another.
`SetReference` and `CheckAndSetReference` call `flush()` first. `flush()`
waits for every queued upload and surfaces the first error. A reference
therefore never goes live while it points at an object whose upload failed or
-is still in flight.
+is still in flight. The first upload error remains fatal for that uploader.
+Failed jobs delete their staging files, so a second flush cannot retry them.
+A later ref update on the same writer therefore returns the same upload error.
+A fresh scoped storer has an independent uploader.
+
+See [the cluster publication check](../usage/pack-publication-check.md) for an
+HTTP and SSH smoke test with a fresh daemon.
`newUploader` sets two bundler limits above their defaults.
@@ -435,9 +441,32 @@ Reads check three tiers, in this order:
2. The pack index.
3. The loose `objects/` key.
-The pack index is built once for each `Storer`, from the `packs/*.cue` files.
+The pack index starts with a listing of `packs/*.cue` for each `Storer`.
Those files are small, and no `.bin` body is read to build it.
+If both the index and loose storage lack an object, the reader lists packs
+again. Concurrent misses share one in-flight refresh. The refresh loads only
+unseen packs and preserves existing tables, iterator snapshots, and local
+registrations. Known objects remain readable during a refresh.
+
+Each completed refresh starts a cooldown for that storer. Misses during the
+cooldown do not list again, and read as absent. The first cooldown is one
+second. A refresh that finds no new pack, or fails, doubles the next cooldown,
+up to ten seconds. A refresh that finds a new pack resets the cooldown to one
+second. Misses never stop future refreshes: a fresh reader that fetches a
+just-pushed ref is the case this path exists for.
+
+A failed refresh is not a storage error for the read. Loose storage already
+answered, so the read returns "object not found". The storer logs the failure
+once per attempt, at warn level. A canceled storer returns the cancellation
+error instead. A failed listing or a corrupt cue on the first, cold build is
+still an error. This bounded recovery does not replace the supported bucket
+consistency requirement.
+
+A storer forgets a pack whose upload failed. If the files of that pack are
+durable in the bucket after all, a later refresh loads it again. A pack that
+this storer registers while a refresh is in flight is never loaded twice.
+
### The pack index (`packtable.go`)
The index is a list of sorted tables, and not a map. A table holds the hashes
diff --git a/docs/architecture/transports.md b/docs/architecture/transports.md
index e3199c3..1202b08 100644
--- a/docs/architecture/transports.md
+++ b/docs/architecture/transports.md
@@ -33,6 +33,18 @@ fixed suffix.
Smart HTTP calls the same go-git server commands with `StatelessRPC: true`.
`GET /info/refs` also sets `AdvertiseRefs: true`.
+For shallow clones, Git first sends a separate wants/deepen request without
+haves or done. The response must end at the shallow-update flush. The upstream
+server goes on to decode haves, which hits EOF, or it appends a NAK. Git
+mistakes that NAK for the next round's shallow list.
+
+`uploadpack.go` holds a fork of go-git's `transport.UploadPack` with one seam.
+If a stateless request carries a deepen, the fork peeks the buffered reader for
+EOF after the first upload-request decode. If the body is exhausted, the fork
+writes the shallow update and returns. A request with haves or done after the
+deepen takes the upstream path. Everything else in the fork mirrors upstream.
+SSH uses the upstream command directly.
+
## SSH (`ssh.go`)
`newSSHServer` builds the gliderlabs/ssh server and the host key. `handleSSH`
diff --git a/docs/plans/2026-10-15-issue-28-pack-visibility.md b/docs/plans/2026-10-15-issue-28-pack-visibility.md
new file mode 100644
index 0000000..45677dd
--- /dev/null
+++ b/docs/plans/2026-10-15-issue-28-pack-visibility.md
@@ -0,0 +1,248 @@
+# Fresh-reader pack visibility: investigation and implementation plan
+
+> For agentic workers: Use `superpowers:executing-plans` for execution. Complete the evidence gate before selecting a production change.
+
+**Goal:** After a successful push, a fresh reader can fetch the advertised commit and its reachable objects.
+
+**Additional requirement:** All bin uploads must finish before hooks execute. Log exactly `waiting for packfiles to finish pushing` before the upload wait and before hook execution.
+
+**Architecture:** Keep the shared Tigris storer as the publication boundary. Complete each `.bin` upload before its `.cue` upload. Complete all uploads before publishing refs. Require a storage consistency contract that preserves this order for readers.
+
+**Tech stack:** Go 1.26, AWS S3 SDK, Tigris, go-git v6, real Git protocol tests.
+
+**Spec:** [Issue 28](https://github.com/tigrisdata/objgit/issues/28), including the requirement in its first comment.
+
+**Status:** Local implementation and checks complete. Production root cause remains unconfirmed. Source review used commit `b60f728b57b24720389f4fc3c5f8fd59b1b89719`.
+
+## Execution decisions
+
+The user approved this plan, then requested a more ambitious implementation.
+The user also stated that the cluster is inaccessible and requested a personal-cluster test procedure.
+
+- A regression test reproduced unsafe ref publication after an earlier upload failure. Upload errors now remain fatal for that writer.
+- The requested wait log precedes the ref-publication barrier and hook execution.
+- Discovery recovery uses a one-second cooldown after each attempt. Concurrent misses share one refresh, and later misses can retry.
+- The refresh adds only unseen immutable packs. It preserves existing tables, snapshots, and local registrations.
+- Real HTTP and SSH tests use the AWS SDK and Tigris storer against a gated S3 HTTP fixture.
+- Those tests reproduce bin/cue failures and prove that hooks wait for successful publication.
+- The new HTTP shallow-clone test exposed an extra NAK in the initial shallow response. A narrow HTTP adapter ends that round at its flush.
+- Production bucket inspection and live-cluster runs remain operator tasks. See [the test procedure](../usage/pack-publication-check.md).
+- No deployment, bucket migration, or production repair was performed.
+
+## Validation results
+
+- `go test ./...`: passed.
+- `go build ./...`: passed.
+- `go test -race ./internal/storage/tigris`: passed.
+- Focused daemon race tests for publication, HTTP/SSH hooks, and shallow response framing: passed.
+- The broader daemon race run reported a race in the unchanged `TestSmartHTTPHealsDanglingHEAD` memory fixture.
+- A review found that refresh blocked known-object reads. A failing regression test reproduced it, and the corrected fast path passes.
+- The live-bucket and personal-cluster tests were not run. The user will run the supplied procedure.
+
+The task checklists below preserve the original investigation plan. Production-only items remain pending operator evidence.
+
+## Findings
+
+- `packJob.run` in `internal/storage/tigris/packwriter.go` already performs synchronous `.bin` and `.cue` PUTs in that order.
+- `commitRefsNotifying` in `refcache.go` flushes all queued uploads before its conditional ref PUT.
+- `ensurePacksBuilt` in `packindex.go` caches a successful listing, including an incomplete listing, for the instance lifetime.
+- `packLookup` does not refresh that listing after a miss. The object reader then tries the loose-object key.
+- Each request gets a scoped storer. A fresh request can recover from temporary visibility lag, but repeated failures need further evidence.
+- Existing tests cover cold pack reads and a general upload failure. They do not isolate `.cue` failure or delayed discovery after successful PUTs.
+- The architecture document requires Single-region or Multi-region buckets. The affected bucket mode and deployed image digest remain unknown.
+
+A local GET after a PUT cannot prove visibility in every reader region. A retry cannot provide a strict publication guarantee either.
+
+## Global constraints
+
+- Preserve the `.cue` format and existing repository keys.
+- Keep pack uploads concurrent across containers. Preserve sequential publication within each container.
+- Keep uploads streamed from staging files and retain the current memory bounds.
+- Use table-driven tests with `tt`, channel barriers, and bounded waits. Do not use sleeps to control races.
+- Keep transport behavior shared through the storer.
+- Require successful completion of all bin and cue uploads before publishing branch tips and executing their hooks.
+- Use `Signed-off-by: Xe Iaso ` for implementation commits.
+- Do not delete or rewrite affected production objects during diagnosis.
+
+## Review focus
+
+1. A failed `.bin` PUT must prevent its `.cue` PUT and every dependent ref update.
+2. A failed `.cue` PUT must prevent refs, even when its `.bin` exists.
+3. A successful but incomplete LIST must be distinguishable from absent or corrupt stored data.
+4. Fresh readers must read complete object bodies without writer overlays or a shared local pack cache.
+5. Cancellation, permission errors, and corruption must remain errors rather than become missing-object retries.
+
+## Task 1: Establish the production failure boundary
+
+**Files:** Record evidence in this plan. Read `docs/architecture/tigris-storer.md` and `docs/reference/tigris-backend.md`.
+
+- [ ] Record the deployment image digest and source revision. Compare its upload and discovery code with the reviewed revision.
+- [ ] Read the bucket mode, endpoint, and writer/reader regions through the Tigris bucket tools.
+- [ ] Read `raidsim/arcadion/packed-refs` and any legacy ref for `refs/heads/main`.
+- [ ] List every page of `raidsim/arcadion/packs/`. Parse the cues to locate `cad12a6e25760bbcce5dac516047cfadeb0d78d9`.
+- [ ] For a matching cue, GET its bin directly. Check the indexed range, reconstructed object hash, and delta bases.
+- [ ] Compare LIST with direct GET for known keys from upload logs. Record timestamps and request identifiers across affected regions.
+- [ ] Classify the result using this decision table.
+
+| Evidence | Next action |
+| ---------------------------------------------------------- | ---------------------------------------------------------------------------------------------------- |
+| Deployed code differs in publication ordering | Reproduce that ordering and deploy the corrected sequential path. |
+| Cue GET succeeds but LIST omits its known key | Investigate discovery visibility and the bucket consistency contract. |
+| Cue exists but bin GET fails | Distinguish failed upload, delayed visibility, and missing data from logs and repeated direct reads. |
+| Cue contains the commit and bin bytes are valid | Trace parsing, lookup, prefix, and hash-format handling with that fixture. |
+| Commit is absent from all available cues and loose storage | Investigate ingestion or lost data. A read retry cannot reconstruct it. |
+| Unsupported consistency mode explains the failure | Use a supported bucket configuration and prevent unsupported writable deployments. |
+
+**Exit criterion:** Identify the failed boundary or explicitly record which production evidence is unavailable. Do not label listing lag as the root cause without evidence.
+
+## Task 2: Pin publication ordering and failure behavior
+
+**Files:** `internal/storage/tigris/client_test.go`, `pack_test.go`, and `refcommit_test.go`.
+
+**Interface:** Add test-only gates and failures keyed by S3 operation and key. Preserve the existing `s3API` interface.
+
+- [ ] Add a channel gate before a selected PUT becomes visible. Invoke gates outside the fake client's mutex.
+- [ ] Record PUT start, completion, and failure events under a mutex.
+- [ ] Add `TestPackPublicationOrder` with blocked bin, failed bin, blocked cue, failed cue, and successful upload cases.
+- [ ] Start a ref update while the bin is blocked. Assert that neither cue publication nor ref publication occurs.
+- [ ] Release the bin and block the cue. Assert that the bin exists and the ref remains unchanged.
+- [ ] Inject each PUT failure separately. Assert the returned error and unchanged remote ref through a fresh storer.
+- [ ] Repeat with several containers. Require all cue uploads to complete before the ref PUT.
+- [ ] Exercise both packed and legacy ref paths where supported by the test constructors.
+- [ ] Check a second ref attempt after a failed flush. The uploader currently clears its error on flush, so record any unsafe reuse.
+
+The required event order for every referenced container is:
+
+```text
+bin PUT succeeds -> cue PUT starts -> cue PUT succeeds -> ref PUT starts
+```
+
+Run:
+
+```sh
+go test ./internal/storage/tigris -run 'TestPackPublicationOrder|TestPackUploadFailureBlocksRef|TestCommitRefs'
+```
+
+Some ordering cases are expected to pass on current code. They protect the explicit issue requirement without inventing a missing fix.
+
+## Task 3: Reproduce discovery failures and select the smallest correction
+
+**Files:** `internal/storage/tigris/client_test.go`, `pack_test.go`, and, only for a reproduced defect, `packindex.go`, `objects.go`, or `upload.go`.
+
+- [ ] Add independent test controls for LIST omission, cue GET absence, and bin GET absence after successful PUTs.
+- [ ] Publish a real pack and ref. Construct a fresh reader with no shared pack cache.
+- [ ] Hide its cue from the first successful listing. Restore listing visibility and read the advertised commit again.
+- [ ] Assert the current behavior: the same reader retains its incomplete index, while another fresh reader can discover the pack.
+- [ ] Add separate cases for permanent absence, permission errors, malformed cues, and canceled contexts.
+- [ ] Select the production correction from Task 1 evidence.
+
+**Preferred correction:** Enforce the existing supported storage contract and preserve publication ordering. Do not add a manifest or global retry loop by default.
+
+**If supported storage exposes a reproducible stale-discovery window:** Add one request-bounded index refresh after both pack and loose lookup miss. Coalesce concurrent refreshes. Preserve local registrations and install the new remote tables atomically. Do not append repeated copies of old remote tables. Keep corruption and permission errors outside this path. Add the corresponding cases for `HasEncodedObject`, `EncodedObjectSize`, and `EncodedObject`.
+
+**If failure is in ingestion or upload:** Correct that reproduced defect instead. Preserve a failed transaction state if the second-flush test proves that refs can bypass an earlier upload error.
+
+**If the bucket mode is unsupported:** Determine whether the provider exposes authoritative bucket-mode metadata to the daemon's credentials. Reject known unsupported modes before enabling writes. If metadata is unavailable, document an explicit operator preflight instead of claiming automatic detection. Define that configuration interface in a follow-up design before coding it.
+
+Retries are bounded recovery only. Supporting eventual consistency as a publication guarantee requires a separate design for ref and pack discovery.
+
+## Task 4: Make the pre-hook upload barrier explicit
+
+**Files:** `cmd/objgitd/receivepack.go`, `cmd/objgitd/hooks.go`, the receive-pack tests, and `docs/architecture/hooks.md`.
+
+**Interface:** Preserve the existing `onUpdated` callback and storage interfaces. Use the upload flush inside reference publication as the barrier.
+
+The current sequence is `writePack -> updateReferences -> report-status -> onUpdated -> snapshots -> hooks -> webhooks`.
+The Tigris ref mutators already wait for uploads. Hooks must remain after that successful wait.
+
+- [ ] Add the following server log before `updateReferences` for a successfully received pack, before its upload flush can block:
+
+```go
+if needPackfile {
+ slog.Info("waiting for packfiles to finish pushing")
+}
+```
+
+- [ ] Keep the exact message text. Add repository context if the existing logger provides it.
+- [ ] Do not place the log only in `runHook`: that location is after the upload wait.
+- [ ] Keep the ref-publication barrier. Do not add a second post-commit flush that can consume or conceal upload failures.
+- [ ] Add a gated upload test through the shared receive-pack path. Block a bin PUT and capture logs and hook execution.
+- [ ] Assert that the exact log appears while the upload is blocked and that the hook has not started.
+- [ ] Release the bin and block the cue. Assert that the hook still has not started.
+- [ ] Release the cue. Assert that the ref update succeeds before the hook starts.
+- [ ] Fail either upload. Assert that the ref stays unchanged and the hook never executes.
+- [ ] Include multiple containers so one completed bin cannot release the hook while another upload remains pending.
+- [ ] Cover HTTP and SSH hook dispatch, plus a deletion-only push that does not emit a pack-upload wait message.
+- [ ] Update the hook architecture document to state the upload, ref publication, and hook execution order.
+
+Expected successful sequence:
+
+```text
+waiting for packfiles to finish pushing
+all bin and cue PUTs complete successfully
+refs commit
+hook executes
+```
+
+The log describes the server upload barrier. It does not claim visibility in every region.
+
+## Task 5: Check the full push-to-fetch contract
+
+**Files:** `internal/storage/tigris/livebucket_test.go`, `cmd/objgitd/http_test.go`, `cmd/objgitd/ssh_test.go`, and `docs/architecture/tigris-storer.md`.
+
+- [ ] Extend the opt-in live pack test to publish a branch ref and resolve it from a new root storer.
+- [ ] Read and hash the commit, trees, blobs, and delta bases reachable from that ref.
+- [ ] Use a unique test prefix. Limit cleanup to keys created by the test.
+- [ ] Run an opt-in daemon smoke test against the supported bucket with a fresh process and empty pack cache.
+- [ ] Push, check `ls-remote`, and clone with `--depth=1 --no-checkout` over HTTP and SSH.
+- [ ] Check the cloned tip and run `git fsck --full` in each shallow clone.
+- [ ] Repeat a fetch after a second push and after a daemon restart.
+- [ ] Update the architecture document with the measured failure boundary, publication contract, and supported deployment configuration.
+
+Run the focused tests first, then:
+
+```sh
+go test -race ./internal/storage/tigris
+go test ./cmd/objgitd -run 'TestSmartHTTP|TestSSH'
+go test ./...
+go build ./...
+```
+
+The full suite needs the Git LFS artifacts described in `AGENTS.md`. Live tests need an explicitly configured test bucket.
+
+## Acceptance criteria
+
+- The diagnosed failure has a regression test that fails before its correction and passes afterward.
+- Ordering tests cover both bin and cue failure independently.
+- Hooks cannot start while any bin or cue upload remains pending or after an upload failure.
+- The exact log `waiting for packfiles to finish pushing` precedes the upload wait and hook execution.
+- A successful push is fetchable from a fresh process on supported storage through HTTP and SSH.
+- The report identifies the production bucket mode and image revision, or clearly states that those facts remain unavailable.
+- No local read probe or finite retry is described as proof of global visibility.
+- Existing missing production data, if found, has a separate recovery proposal based on an intact source repository.
+
+## Review follow-up
+
+The review found two limits in the first implementation. This follow-up replaces both mechanisms.
+
+1. Replace the HTTP request decode/re-encode pass with a packet reader. Forward the original bytes and retain at most one packet. Let upstream validate the request once.
+2. Replace the lifetime refresh budget with a one-second cooldown after each completed attempt. Concurrent misses share one result. Errors remain visible during the cooldown. Later misses can retry, including after an earlier genuine miss or storage failure.
+3. Test byte preservation, fragmented input, shallow-round detection, repeated publications, failure recovery, and concurrent requests. Use virtual time for timing-only tests. Full pack fixtures use the real cooldown because the shared compression pools cannot cross virtual-time boundaries.
+4. Run the full suite, build, and focused race checks. Request another skeptical review before completion.
+
+This policy bounds refresh frequency per storer. It does not promise immediate visibility during the cooldown or change the bucket consistency requirement.
+
+Follow-up validation:
+
+- The old refresh implementation fails both new recovery cases: a genuine missing probe and a transient listing error.
+- The corrected reader discovers two successive publications through all three object lookup APIs. Concurrent misses share each attempt.
+- Virtual-time tests cover a slow refresh, cooldown timing, cancellation during cooldown, and cancellation of an active waiter.
+- HTTP tests cover fragmented input, maximum-size packets, malformed packets, read errors, and exact request bytes.
+- The adapter returns its packet buffer after the first flush, before pack generation.
+- `go test ./...` and `go build ./...` pass.
+- The full storage race suite and focused daemon race checks pass. Final focused race checks include all new cases.
+- A second skeptical review found no blockers. Its three optional improvements are included.
+
+The adapter benchmark compares packet forwarding with the previous decode/replay pass. In the final local sample, 10,000 wants took 0.25 ms versus 1.55 ms. Allocations were about 4.3 KB versus 3.66 MB. These measurements cover adapter overhead, not total fetch throughput. Run `go test ./cmd/objgitd -run '^$' -bench BenchmarkUploadRequest -benchmem` to repeat them.
+
+Production-cluster validation remains pending. The operator procedure still applies.
diff --git a/docs/usage/pack-publication-check.md b/docs/usage/pack-publication-check.md
new file mode 100644
index 0000000..b6211c3
--- /dev/null
+++ b/docs/usage/pack-publication-check.md
@@ -0,0 +1,171 @@
+# Check pack publication on a personal cluster
+
+This procedure checks issue 28 with the patched image. It uses a new repository in your test bucket.
+The local regression tests also check blocked and failed uploads. A successful cluster smoke test alone cannot prove behavior during every failure.
+
+## 1. Record the deployment and bucket
+
+Set these values for your personal cluster:
+
+```sh
+export OBJGIT_CONTEXT='your-context'
+export OBJGIT_NAMESPACE='objgit'
+export OBJGIT_DEPLOYMENT='objgitd'
+export OBJGIT_CONTAINER='objgitd'
+export OBJGIT_BUCKET='your-test-bucket'
+export OBJGIT_IMAGE='your-registry/objgit:issue-28-test'
+```
+
+Record the existing image and bucket configuration:
+
+```sh
+kubectl --context "$OBJGIT_CONTEXT" -n "$OBJGIT_NAMESPACE" get deployment "$OBJGIT_DEPLOYMENT" \
+ -o jsonpath='{.spec.template.spec.containers[*].image}{"\n"}'
+tigris buckets get "$OBJGIT_BUCKET" --json
+```
+
+The storer requires a Single-region or Multi-region bucket. Record the pod region and bucket locations.
+An unsupported bucket mode needs a separate deployment correction. These code changes do not provide global read consistency.
+
+## 2. Build and install the test image
+
+Run these commands from the patched checkout. Select the platform used by your cluster nodes.
+
+```sh
+git lfs pull
+docker buildx build --platform linux/amd64 -t "$OBJGIT_IMAGE" --push .
+kubectl --context "$OBJGIT_CONTEXT" -n "$OBJGIT_NAMESPACE" set image \
+ deployment/"$OBJGIT_DEPLOYMENT" "$OBJGIT_CONTAINER=$OBJGIT_IMAGE"
+```
+
+For this test deployment, enable hooks and both transports. Disable the pack cache to force reads from the bucket.
+These commands change your test deployment's configuration:
+
+```sh
+kubectl --context "$OBJGIT_CONTEXT" -n "$OBJGIT_NAMESPACE" set env \
+ deployment/"$OBJGIT_DEPLOYMENT" \
+ BUCKET="$OBJGIT_BUCKET" ALLOW_PUSH=true ALLOW_HOOKS=true \
+ HTTP_BIND=:8080 SSH_BIND=:2222 PACK_CACHE_BYTES=0
+kubectl --context "$OBJGIT_CONTEXT" -n "$OBJGIT_NAMESPACE" rollout status \
+ deployment/"$OBJGIT_DEPLOYMENT" --timeout=180s
+kubectl --context "$OBJGIT_CONTEXT" -n "$OBJGIT_NAMESPACE" get pods \
+ -l app.kubernetes.io/name=objgitd \
+ -o 'custom-columns=NAME:.metadata.name,IMAGE:.spec.containers[0].image,IMAGEID:.status.containerStatuses[0].imageID'
+```
+
+Retain the existing credential configuration. No credential values belong in the test report.
+
+In another terminal, forward both ports:
+
+```sh
+kubectl --context "$OBJGIT_CONTEXT" -n "$OBJGIT_NAMESPACE" port-forward \
+ deployment/"$OBJGIT_DEPLOYMENT" 18080:8080 12222:2222
+```
+
+## 3. Push a repository with a hook
+
+Use Bash or Zsh for these commands:
+
+```sh
+export OBJGIT_TEST_ROOT="$(mktemp -d)"
+export OBJGIT_TEST_REPO="smoke/pack-publication-$(date +%s)"
+export OBJGIT_HTTP="http://127.0.0.1:18080/$OBJGIT_TEST_REPO.git"
+export OBJGIT_SSH="ssh://git@127.0.0.1:12222/$OBJGIT_TEST_REPO.git"
+export GIT_SSH_COMMAND="ssh -o StrictHostKeyChecking=accept-new -o UserKnownHostsFile=$OBJGIT_TEST_ROOT/known_hosts"
+
+git init -b main "$OBJGIT_TEST_ROOT/source"
+git -C "$OBJGIT_TEST_ROOT/source" config user.name 'Publication test'
+git -C "$OBJGIT_TEST_ROOT/source" config user.email 'publication-test@example.invalid'
+mkdir -p "$OBJGIT_TEST_ROOT/source/.objgit/hooks"
+printf 'echo publication-hook-ran\n' > "$OBJGIT_TEST_ROOT/source/.objgit/hooks/receive-pack"
+dd if=/dev/urandom of="$OBJGIT_TEST_ROOT/source/payload.bin" bs=1048576 count=16
+
+git -C "$OBJGIT_TEST_ROOT/source" add .
+git -C "$OBJGIT_TEST_ROOT/source" commit -m 'test pack publication'
+git -C "$OBJGIT_TEST_ROOT/source" push "$OBJGIT_HTTP" main
+```
+
+The push output must contain `publication-hook-ran`. The server log must contain these messages in order:
+
+```text
+hook: running
+hook: finished
+```
+
+Capture the log before restarting the pod:
+
+```sh
+kubectl --context "$OBJGIT_CONTEXT" -n "$OBJGIT_NAMESPACE" logs \
+ deployment/"$OBJGIT_DEPLOYMENT" -c "$OBJGIT_CONTAINER" --since=10m \
+ > "$OBJGIT_TEST_ROOT/server-before-restart.log"
+```
+
+The server publishes a ref only after the storer flushes every queued bin and cue upload.
+The hook starts after that ref update. This smoke test does not show that order.
+`TestTigrisPushPublication` in `cmd/objgitd/s3_publication_test.go` holds the uploads open and checks that order.
+To check that order on a workstation, run `go test -run TestTigrisPushPublication ./cmd/objgitd/`.
+
+## 4. Fetch through fresh readers
+
+First clone immediately after the push:
+
+```sh
+git ls-remote "$OBJGIT_HTTP" refs/heads/main
+git clone --depth=1 --no-checkout "$OBJGIT_HTTP" "$OBJGIT_TEST_ROOT/http"
+git clone --depth=1 --no-checkout "$OBJGIT_SSH" "$OBJGIT_TEST_ROOT/ssh"
+test "$(git -C "$OBJGIT_TEST_ROOT/source" rev-parse HEAD)" = "$(git -C "$OBJGIT_TEST_ROOT/http" rev-parse HEAD)"
+test "$(git -C "$OBJGIT_TEST_ROOT/source" rev-parse HEAD)" = "$(git -C "$OBJGIT_TEST_ROOT/ssh" rev-parse HEAD)"
+git -C "$OBJGIT_TEST_ROOT/http" fsck --full
+git -C "$OBJGIT_TEST_ROOT/ssh" fsck --full
+```
+
+Then restart the daemon:
+
+```sh
+kubectl --context "$OBJGIT_CONTEXT" -n "$OBJGIT_NAMESPACE" rollout restart deployment/"$OBJGIT_DEPLOYMENT"
+kubectl --context "$OBJGIT_CONTEXT" -n "$OBJGIT_NAMESPACE" rollout status deployment/"$OBJGIT_DEPLOYMENT" --timeout=180s
+```
+
+Restart the port-forward command after the old pod exits. Repeat both clones into new directories, such as `http-restart` and `ssh-restart`.
+Compare their tips and run `git fsck --full` again.
+
+Exercise a second push through SSH:
+
+```sh
+printf 'second push\n' > "$OBJGIT_TEST_ROOT/source/second.txt"
+git -C "$OBJGIT_TEST_ROOT/source" add second.txt
+git -C "$OBJGIT_TEST_ROOT/source" commit -m 'test second publication'
+git -C "$OBJGIT_TEST_ROOT/source" push "$OBJGIT_SSH" main
+git -C "$OBJGIT_TEST_ROOT/http" fetch --depth=1 origin main
+git -C "$OBJGIT_TEST_ROOT/ssh" fetch --depth=1 origin main
+test "$(git -C "$OBJGIT_TEST_ROOT/source" rev-parse HEAD)" = "$(git -C "$OBJGIT_TEST_ROOT/http" rev-parse FETCH_HEAD)"
+test "$(git -C "$OBJGIT_TEST_ROOT/source" rev-parse HEAD)" = "$(git -C "$OBJGIT_TEST_ROOT/ssh" rev-parse FETCH_HEAD)"
+```
+
+## 5. Run the live storage test
+
+From a machine with the test bucket's credentials, run:
+
+```sh
+OBJGIT_TIGRIS_LIVE_BUCKET="$OBJGIT_BUCKET" \
+ go test ./internal/storage/tigris -run '^TestLiveBucketPackRoundTrip$' -count=1 -v
+```
+
+This test writes a unique prefix, publishes a ref, and reads every fixture object through a new root storer.
+It hashes the complete object bodies and removes its own keys after the run.
+The manual smoke repository remains in the bucket for inspection.
+
+## If a clone fails
+
+Preserve the server logs, image digest, bucket mode, region, advertised SHA, transport, and timestamps.
+Record whether a new clone succeeds later without another push.
+
+Use the existing object tools to compare LIST and direct GET for the known pack keys from upload diagnostics:
+
+- A readable cue omitted from LIST indicates discovery visibility trouble.
+- A readable cue with an unreadable bin indicates upload or bin visibility trouble.
+- A missing commit in all cues requires investigation of ingestion or lost data.
+- A valid cue and bin require investigation of lookup, prefix, hash format, or delta resolution.
+
+Do not rewrite the affected ref or delete packs before collecting this evidence.
+Restore your previous test image and configuration after the investigation.
diff --git a/internal/storage/tigris/client_test.go b/internal/storage/tigris/client_test.go
index 669d330..87ceed6 100644
--- a/internal/storage/tigris/client_test.go
+++ b/internal/storage/tigris/client_test.go
@@ -51,6 +51,8 @@ type fakeS3 struct {
delErr error
listErr error
+ beforePut func(string) error // test gate, called outside mu before each PUT
+
putDelay time.Duration // artificial latency before PutObject lands, for async-upload tests
// putHook fires inside PutObject after the injected-error check and before
@@ -408,6 +410,11 @@ func parseByteRange(header string, size int) (start, end int, ok bool) {
}
func (f *fakeS3) PutObject(_ context.Context, p *s3.PutObjectInput, _ ...func(*s3.Options)) (*s3.PutObjectOutput, error) {
+ if f.beforePut != nil {
+ if err := f.beforePut(sv(p.Key)); err != nil {
+ return nil, err
+ }
+ }
if f.putDelay > 0 {
// Held outside the lock, so overlapping uploads really do overlap and
// the high-water mark means something.
diff --git a/internal/storage/tigris/livebucket_test.go b/internal/storage/tigris/livebucket_test.go
index c32fbfb..86c0d29 100644
--- a/internal/storage/tigris/livebucket_test.go
+++ b/internal/storage/tigris/livebucket_test.go
@@ -161,8 +161,52 @@ func TestLiveBucketPackRoundTrip(t *testing.T) {
}
})
- // Fresh Storer: cold cue-index build, then reads straight off the wire.
- reader := base.Scoped(prefix)
+ // Publish a commit and resolve it through a new root, with no writer
+ // overlay or shared local cache.
+ var tip plumbing.Hash
+ for _, h := range fx.hashes {
+ if fx.byHash[h].typ == plumbing.CommitObject {
+ tip = h
+ break
+ }
+ }
+ if tip.IsZero() {
+ t.Fatal("fixture contains no commit")
+ }
+ ref := plumbing.NewHashReference(plumbing.NewBranchReferenceName("main"), tip)
+ if err := s.SetReference(ref); err != nil {
+ t.Fatalf("publish ref: %v", err)
+ }
+ fresh, err := New(ctx, bucket)
+ if err != nil {
+ t.Fatal(err)
+ }
+ reader := fresh.Scoped(prefix)
+ gotRef, err := reader.Reference(ref.Name())
+ if err != nil || gotRef.Hash() != tip {
+ t.Fatalf("cold ref: %v, %v", gotRef, err)
+ }
+ for _, h := range fx.hashes {
+ obj, err := reader.EncodedObject(plumbing.AnyObject, h)
+ if err != nil {
+ t.Fatalf("cold object %s: %v", h, err)
+ }
+ r, err := obj.Reader()
+ if err != nil {
+ t.Fatal(err)
+ }
+ hasher := plumbing.NewHasher(reader.of, obj.Type(), obj.Size())
+ _, err = io.Copy(hasher, r)
+ r.Close()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if hasher.Sum() != h {
+ t.Fatalf("cold object hash differs for %s", h)
+ }
+ }
+ // Use another cold reader to exercise ranged GET and prefetch below.
+ reader = fresh.Scoped(prefix)
blobs := blobHashes(fx)
if len(blobs) < 4 {
diff --git a/internal/storage/tigris/objects.go b/internal/storage/tigris/objects.go
index 224f5d7..6e71bc5 100644
--- a/internal/storage/tigris/objects.go
+++ b/internal/storage/tigris/objects.go
@@ -48,7 +48,11 @@ func (s *Storer) headInfo(h plumbing.Hash) (objHead, error) {
switch {
case err == nil:
case isNotFound(err):
- return zeroObjHead, plumbing.ErrObjectNotFound
+ e, err := s.lookupAfterMiss(h)
+ if err != nil {
+ return zeroObjHead, err
+ }
+ return objHead{typ: e.typ, size: e.raw}, nil
default:
return zeroObjHead, fmt.Errorf("tigris: head %s: %w", h.String(), err)
}
@@ -56,6 +60,35 @@ func (s *Storer) headInfo(h plumbing.Hash) (objHead, error) {
return parseObjHead(h, out.Metadata, out.ContentLength)
}
+// lookupAfterMiss is a read's last step once loose storage has said 404 for
+// h: the pack index may predate a pack another writer just published, so it
+// is refreshed and searched once more. The result is the entry, or
+// ErrObjectNotFound.
+//
+// A refresh that fails does not make this read a storage error. Loose storage
+// has already answered, and a missing object is a normal answer: during a
+// fetch negotiation go-git walks every `have` the client sends, tolerates
+// only ErrObjectNotFound for one this server lacks, and fails the whole
+// upload-pack on anything else. refreshPacks logs the failure once per
+// attempt and rate-limits the next one, so a sick bucket is neither hidden
+// nor hammered. Cancellation is the one exception: a canceled storer reports
+// that, so a read does not pretend to have searched.
+func (s *Storer) lookupAfterMiss(h plumbing.Hash) (packEntry, error) {
+ if err := s.refreshPacks(); err != nil {
+ if cerr := s.ctx.Err(); cerr != nil {
+ return packEntry{}, cerr
+ }
+ }
+ e, ok, err := s.packLookup(h)
+ if err != nil {
+ return packEntry{}, err
+ }
+ if !ok {
+ return packEntry{}, plumbing.ErrObjectNotFound
+ }
+ return e, nil
+}
+
// parsePendingHead mirrors parseObjHead for a pending-overlay entry, whose
// type/size were captured straight from the writer rather than S3 metadata.
func parsePendingHead(h plumbing.Hash, p pendingMeta) (objHead, error) {
@@ -144,7 +177,11 @@ func (s *Storer) EncodedObject(t plumbing.ObjectType, h plumbing.Hash) (plumbing
switch {
case err == nil:
case isNotFound(err):
- return nil, plumbing.ErrObjectNotFound
+ e, err := s.lookupAfterMiss(h)
+ if err != nil {
+ return nil, err
+ }
+ return s.packObject(t, h, e)
default:
return nil, fmt.Errorf("tigris: get %s: %w", h.String(), err)
}
diff --git a/internal/storage/tigris/packindex.go b/internal/storage/tigris/packindex.go
index 912ee01..8dbc73c 100644
--- a/internal/storage/tigris/packindex.go
+++ b/internal/storage/tigris/packindex.go
@@ -423,9 +423,20 @@ type packAccess struct {
// build.
type packIndex struct {
mu sync.Mutex
+ buildMu sync.Mutex // serializes cold builds and the bounded refresh
+ indexed map[string]struct{} // immutable packs already registered or loaded
built bool
hashSize int
+ // The refresh fields are under refreshMu, not mu. See refreshPacks.
+ refreshMu sync.Mutex
+ refresh *packRefresh // the attempt in flight, if any
+ refreshAfter time.Time // no attempt starts before this
+ refreshWait time.Duration // the cooldown the last attempt started
+ refreshBase time.Duration // the cooldown after an attempt that found a pack
+ refreshMax time.Duration // the cap on a doubled cooldown
+ now func() time.Time
+
ids []string // pack number -> pack id
idNum map[string]uint32 // pack id -> pack number
tables []*packTable // oldest first; see maxPackTables
@@ -444,13 +455,25 @@ type packIndex struct {
// them all into one.
const maxPackTables = 16
+// The refresh cooldown starts at refreshBaseCooldown. Each attempt that finds
+// no new pack doubles it, up to refreshMaxCooldown; an attempt that finds one
+// resets it. See refreshPacks.
+const (
+ refreshBaseCooldown = time.Second
+ refreshMaxCooldown = 10 * time.Second
+)
+
func newPackIndex(hashSize int) *packIndex {
return &packIndex{
- hashSize: hashSize,
- idNum: make(map[string]uint32),
- sizes: make(map[string]int64),
- local: make(map[string]string),
- access: make(map[string]*packAccess),
+ hashSize: hashSize,
+ indexed: make(map[string]struct{}),
+ idNum: make(map[string]uint32),
+ sizes: make(map[string]int64),
+ local: make(map[string]string),
+ access: make(map[string]*packAccess),
+ refreshBase: refreshBaseCooldown,
+ refreshMax: refreshMaxCooldown,
+ now: time.Now,
}
}
@@ -491,18 +514,26 @@ func (p *packIndex) register(id string, recs []cueRecord, localBin string) {
p.mu.Lock()
defer p.mu.Unlock()
num := p.packNum(id)
- if _, ok := p.dead[num]; ok {
- // A pack that failed once and is written again is live. The map is
- // shared with snapshots, so replace it.
- dead := maps.Clone(p.dead)
- delete(dead, num)
- p.dead = dead
- }
+ p.indexed[id] = struct{}{}
+ p.revive(num)
p.addTable(newPackTable(p.hashSize, num, recs))
p.sizes[id] = packExtent(recs)
p.local[id] = localBin
}
+// revive takes pack num off the dead set, for a pack that failed once and is
+// live again: written again, or found durable in the bucket after all. The
+// map is shared with snapshots, so it is replaced and never edited. Callers
+// hold p.mu.
+func (p *packIndex) revive(num uint32) {
+ if _, ok := p.dead[num]; !ok {
+ return
+ }
+ dead := maps.Clone(p.dead)
+ delete(dead, num)
+ p.dead = dead
+}
+
// markUploaded drops a pack's local-cache visibility once its upload lands:
// its entries stay valid (now served from S3 instead), only the local .bin
// path stops being offered to readers.
@@ -514,9 +545,14 @@ func (p *packIndex) markUploaded(id string) {
// deregister removes a pack that failed to upload entirely: its entries must
// not point at a pack S3 will never have.
+//
+// It also forgets the pack in indexed. An upload can report failure after the
+// bucket has the bytes, when only the reply was lost, and a later refresh must
+// then be free to load the pack from its .cue like any other.
func (p *packIndex) deregister(id string) {
p.mu.Lock()
defer p.mu.Unlock()
+ delete(p.indexed, id)
if num, ok := p.idNum[id]; ok {
// A new slice and a new map, never edits in place: a packSnapshot
// shares both.
@@ -656,11 +692,8 @@ func (p *packIndex) bulkCopy(id string) (*os.File, bool) {
}
}
-// ensurePacksBuilt lists packs/*.cue under this Storer's prefix and merges
-// their entries into the index, once per instance. Not sticky on error — a
-// transient S3 failure is retried on the next call, not remembered forever.
-// Entries this instance has itself register-ed (its own in-flight pushes)
-// are untouched: they live outside anything a listing could ever race with.
+// ensurePacksBuilt loads the initial pack listing. A miss in both pack and
+// loose storage can request another listing through refreshPacks.
func (s *Storer) ensurePacksBuilt() error {
s.packs.mu.Lock()
built := s.packs.built
@@ -668,63 +701,181 @@ func (s *Storer) ensurePacksBuilt() error {
if built {
return nil
}
+ _, err := s.buildPacks(false)
+ return err
+}
+
+// packRefresh holds one attempt's result for all callers that joined it.
+// Closing done publishes err without holding the index or refresh locks.
+type packRefresh struct {
+ done chan struct{}
+ err error
+}
+
+// refreshPacks lists packs/*.cue again, so a reader that built its index
+// before another writer published can still find the new pack. Concurrent
+// misses join one attempt. After an attempt completes, no new attempt starts
+// for one cooldown; a miss inside it returns nil at once, and the index stays
+// as it is.
+//
+// The cooldown adapts. A negotiation can send hundreds of haves this server
+// does not hold, and each one is a miss, so an attempt that found nothing, or
+// failed, doubles the next cooldown up to refreshMax. An attempt that found a
+// pack resets it to refreshBase. Misses never stop refreshing altogether: a
+// fresh reader fetching a just-pushed ref is the case this path exists for.
+//
+// A failed attempt is logged here, once, and its error goes to every caller
+// that joined it. What a read makes of that error is lookupAfterMiss's call.
+func (s *Storer) refreshPacks() error {
+ if err := s.ctx.Err(); err != nil {
+ return err
+ }
+ p := s.packs
+ p.refreshMu.Lock()
+ if active := p.refresh; active != nil {
+ p.refreshMu.Unlock()
+ select {
+ case <-active.done:
+ return active.err
+ case <-s.ctx.Done():
+ return s.ctx.Err()
+ }
+ }
+ if p.now().Before(p.refreshAfter) {
+ p.refreshMu.Unlock()
+ return nil
+ }
+ active := &packRefresh{done: make(chan struct{})}
+ p.refresh = active
+ p.refreshMu.Unlock()
+
+ found, err := s.buildPacks(true)
+ active.err = err
+
+ p.refreshMu.Lock()
+ wait := p.refreshBase
+ if err != nil || found == 0 {
+ wait = min(max(2*p.refreshWait, p.refreshBase), p.refreshMax)
+ }
+ p.refreshWait = wait
+ p.refreshAfter = p.now().Add(wait)
+ p.refresh = nil
+ close(active.done)
+ p.refreshMu.Unlock()
+
+ if err != nil && s.ctx.Err() == nil {
+ slog.Warn("pack refresh failed; misses read as absent until the next attempt",
+ "prefix", s.prefix, "cooldown", wait, "err", err)
+ }
+ return err
+}
+
+// buildPacks lists packs/*.cue and loads every pack the index does not hold
+// yet. With refresh false it is the cold build, which runs once; with refresh
+// true it is a later pass for packs published since. It reports how many
+// packs it added.
+//
+// This storer's own write path can register a pack while the listing is in
+// flight, and that pack's .cue can land in the bucket during the same listing.
+// indexed is checked under mu right before each .cue is fetched, and once more
+// when the tables go in, so no pack is ever loaded twice. The second check
+// can drop a pack from the merged table, which costs one more merge; the race
+// is rare.
+func (s *Storer) buildPacks(refresh bool) (int, error) {
+ s.packs.buildMu.Lock()
+ defer s.packs.buildMu.Unlock()
+ s.packs.mu.Lock()
+ built := s.packs.built
+ s.packs.mu.Unlock()
+ if built && !refresh {
+ return 0, nil
+ }
+ if err := s.ctx.Err(); err != nil {
+ return 0, err
+ }
keys, err := s.listKeys(s.prefix + packPrefix)
if err != nil {
- return fmt.Errorf("tigris: list packs: %w", err)
+ return 0, fmt.Errorf("tigris: list packs: %w", err)
}
// Parse each .cue into a table of its own, and drop its records at once,
// so the cold build never holds a whole repository of cueRecords.
hashLen := s.oh.Size()
- var (
- tables []*packTable
- sizes = make(map[string]int64)
- )
+ type loadedPack struct {
+ id string
+ num uint32
+ table *packTable
+ size int64
+ }
+ var packs []loadedPack
for _, k := range keys {
if !strings.HasSuffix(k, cueSuffix) {
continue
}
id := strings.TrimSuffix(strings.TrimPrefix(k, s.prefix+packPrefix), cueSuffix)
+ s.packs.mu.Lock()
+ _, known := s.packs.indexed[id]
+ s.packs.mu.Unlock()
+ if known {
+ continue
+ }
raw, err := s.fetchSmall(k)
if err != nil {
- return fmt.Errorf("tigris: fetch cue %s: %w", k, err)
+ return 0, fmt.Errorf("tigris: fetch cue %s: %w", k, err)
}
recs, err := parseCue(hashLen, raw)
if err != nil {
- return fmt.Errorf("tigris: parse cue %s: %w", k, err)
+ return 0, fmt.Errorf("tigris: parse cue %s: %w", k, err)
}
s.packs.mu.Lock()
num := s.packs.packNum(id)
s.packs.mu.Unlock()
- tables = append(tables, newPackTable(hashLen, num, recs))
- sizes[id] = packExtent(recs)
+ packs = append(packs, loadedPack{id, num, newPackTable(hashLen, num, recs), packExtent(recs)})
}
// One table for the whole repository makes each lookup one binary search.
- if len(tables) > 1 {
- tables = []*packTable{mergeTables(hashLen, tables, nil)}
+ // Merge before taking mu, so a large merge never holds up a lookup.
+ merge := func(packs []loadedPack) []*packTable {
+ tables := make([]*packTable, 0, len(packs))
+ for _, pk := range packs {
+ tables = append(tables, pk.table)
+ }
+ if len(tables) > 1 {
+ tables = []*packTable{mergeTables(hashLen, tables, nil)}
+ }
+ return tables
}
+ tables := merge(packs)
s.packs.mu.Lock()
defer s.packs.mu.Unlock()
- if s.packs.built {
- return nil // lost a race with a concurrent cold build on this instance
+ // Drop any pack that register added since its .cue was fetched.
+ kept := packs[:0]
+ for _, pk := range packs {
+ if _, ok := s.packs.indexed[pk.id]; !ok {
+ kept = append(kept, pk)
+ }
+ }
+ if len(kept) != len(packs) {
+ tables = merge(kept)
+ }
+ for _, pk := range kept {
+ s.packs.indexed[pk.id] = struct{}{}
+ s.packs.revive(pk.num)
+ if _, ok := s.packs.sizes[pk.id]; !ok {
+ s.packs.sizes[pk.id] = pk.size
+ }
}
// The cold tables go first, which makes them the oldest. The packs this
// instance registered itself are newer, so their records win.
s.packs.tables = append(tables, s.packs.tables...)
- for id, n := range sizes {
- if _, ok := s.packs.sizes[id]; !ok {
- s.packs.sizes[id] = n
- }
- }
if len(s.packs.tables) > maxPackTables {
s.packs.tables = []*packTable{mergeTables(hashLen, s.packs.tables, s.packs.dead)}
s.packs.dead = nil
}
s.packs.built = true
- return nil
+ return len(kept), nil
}
// packLookup reports whether h lives in a pack this Storer knows about,
diff --git a/internal/storage/tigris/publication_test.go b/internal/storage/tigris/publication_test.go
new file mode 100644
index 0000000..679796f
--- /dev/null
+++ b/internal/storage/tigris/publication_test.go
@@ -0,0 +1,168 @@
+package tigris
+
+import (
+ "errors"
+ "io"
+ "os/exec"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/go-git/go-git/v6/plumbing"
+)
+
+// A failed upload cannot be retried: done removes its staging files. Every
+// later ref mutation on that writer must keep returning the upload failure.
+func TestPackUploadFailureRemainsFatal(t *testing.T) {
+ if _, err := exec.LookPath("git"); err != nil {
+ t.Skip("git not installed")
+ }
+ for _, tt := range []struct {
+ name, suffix string
+ packed bool
+ }{
+ {"loose bin", binSuffix, false}, {"loose cue", cueSuffix, false},
+ {"packed bin", binSuffix, true}, {"packed cue", cueSuffix, true},
+ } {
+ t.Run(tt.name, func(t *testing.T) {
+ fx := buildPackFixture(t, 3, false)
+ f := newFakeS3(t)
+ uploadErr := errors.New("injected upload failure")
+ f.beforePut = func(key string) error {
+ if strings.HasSuffix(key, tt.suffix) {
+ return uploadErr
+ }
+ return nil
+ }
+ s := newTestStorer(t, f, WithPackedRefs(tt.packed))
+ writePack(t, s, fx)
+ for _, branch := range []string{"main", "other"} {
+ ref := plumbing.NewHashReference(plumbing.NewBranchReferenceName(branch), fx.hashes[0])
+ if err := s.SetReference(ref); !errors.Is(err, uploadErr) {
+ t.Errorf("SetReference(%s) = %v, want upload error", branch, err)
+ }
+ reader := newTestStorer(t, f, WithPackedRefs(tt.packed))
+ if _, err := reader.Reference(ref.Name()); !errors.Is(err, plumbing.ErrReferenceNotFound) {
+ t.Errorf("failed upload published %s: %v", branch, err)
+ }
+ }
+ })
+ }
+}
+
+func TestPackPublicationOrder(t *testing.T) {
+ if _, err := exec.LookPath("git"); err != nil {
+ t.Skip("git not installed")
+ }
+ for _, tt := range []struct {
+ name string
+ packed bool
+ cap int64
+ }{
+ {"loose single", false, maxPackBytes}, {"packed single", true, maxPackBytes},
+ {"loose multiple", false, 1}, {"packed multiple", true, 1},
+ } {
+ t.Run(tt.name, func(t *testing.T) {
+ fx := buildPackFixture(t, 3, false)
+ f := newFakeS3(t)
+ bins, cues := make(chan struct{}), make(chan struct{})
+ binStarted, cueStarted := make(chan struct{}, 100), make(chan struct{}, 100)
+ var releaseBins, releaseCues sync.Once
+ unblockBins := func() { releaseBins.Do(func() { close(bins) }) }
+ unblockCues := func() { releaseCues.Do(func() { close(cues) }) }
+ defer unblockBins()
+ defer unblockCues()
+ var binCount, cueCount atomic.Int32
+ f.beforePut = func(key string) error {
+ switch {
+ case strings.HasSuffix(key, binSuffix):
+ binCount.Add(1)
+ binStarted <- struct{}{}
+ <-bins
+ case strings.HasSuffix(key, cueSuffix):
+ f.mu.Lock()
+ _, exists := f.objs[strings.TrimSuffix(key, cueSuffix)+binSuffix]
+ f.mu.Unlock()
+ if !exists {
+ t.Error("cue PUT started before its bin completed")
+ }
+ cueCount.Add(1)
+ cueStarted <- struct{}{}
+ <-cues
+ }
+ return nil
+ }
+ s := newTestStorer(t, f, WithPackedRefs(tt.packed), withMaxPackBytes(tt.cap))
+ writePack(t, s, fx)
+ ref := plumbing.NewHashReference(plumbing.NewBranchReferenceName("main"), fx.hashes[0])
+ done := make(chan error, 1)
+ go func() { done <- s.SetReference(ref) }()
+ waitPublication(t, binStarted)
+ assertPending := func() {
+ t.Helper()
+ select {
+ case err := <-done:
+ t.Fatalf("ref update finished before uploads: %v", err)
+ default:
+ }
+ reader := newTestStorer(t, f, WithPackedRefs(tt.packed))
+ if _, err := reader.Reference(ref.Name()); !errors.Is(err, plumbing.ErrReferenceNotFound) {
+ t.Fatalf("ref visible while upload blocked: %v", err)
+ }
+ }
+ assertPending()
+ if cueCount.Load() != 0 {
+ t.Fatal("cue started while bins blocked")
+ }
+ unblockBins()
+ waitPublication(t, cueStarted)
+ assertPending()
+ unblockCues()
+ select {
+ case err := <-done:
+ if err != nil {
+ t.Fatal(err)
+ }
+ case <-time.After(10 * time.Second):
+ t.Fatal("ref update did not finish")
+ }
+ if binCount.Load() != cueCount.Load() {
+ t.Fatalf("bins=%d cues=%d", binCount.Load(), cueCount.Load())
+ }
+ if tt.cap == 1 && binCount.Load() < 2 {
+ t.Fatal("fixture did not split containers")
+ }
+ reader := newTestStorer(t, f, WithPackedRefs(tt.packed))
+ for _, h := range fx.hashes {
+ obj, err := reader.EncodedObject(plumbing.AnyObject, h)
+ if err != nil {
+ t.Fatal(err)
+ }
+ r, err := obj.Reader()
+ if err != nil {
+ t.Fatal(err)
+ }
+ hasher := plumbing.NewHasher(reader.of, obj.Type(), obj.Size())
+ _, err = io.Copy(hasher, r)
+ r.Close()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if hasher.Sum() != h {
+ t.Fatalf("cold read hash differs for %s", h)
+ }
+ }
+ })
+ }
+}
+
+func waitPublication(t *testing.T, ch <-chan struct{}) {
+ t.Helper()
+ select {
+ case <-ch:
+ case <-time.After(10 * time.Second):
+ t.Fatal("upload did not reach gate")
+ }
+}
diff --git a/internal/storage/tigris/tigris.go b/internal/storage/tigris/tigris.go
index f657a10..c5bcbd0 100644
--- a/internal/storage/tigris/tigris.go
+++ b/internal/storage/tigris/tigris.go
@@ -235,6 +235,12 @@ func WithRefCASObserver(fn func()) Option {
return func(s *Storer) { s.refCASObserver = fn }
}
+// WithS3Client uses a configured S3 client instead of creating the default
+// Tigris client. This supports custom endpoints and transport-level tests.
+func WithS3Client(client *s3.Client) Option {
+ return func(s *Storer) { s.client = client }
+}
+
func withClient(c s3API) Option {
return func(s *Storer) { s.client = c }
}
diff --git a/internal/storage/tigris/upload.go b/internal/storage/tigris/upload.go
index caac6c0..096e2ff 100644
--- a/internal/storage/tigris/upload.go
+++ b/internal/storage/tigris/upload.go
@@ -132,7 +132,7 @@ type uploader struct {
b *bundler.Bundler[uploadJob]
mu sync.Mutex
- err error // first upload error since the last flush
+ err error // first upload error; fatal for this uploader
pending map[plumbing.Hash]pendingMeta
}
@@ -244,7 +244,8 @@ func (u *uploader) sizeHint(job uploadJob) int {
}
// flush waits for every upload queued so far to finish and returns the first
-// error any of them hit, clearing it so the next flush starts fresh. Callers
+// error any of them hit. Failed jobs discard their staging files, so the
+// error remains fatal for this uploader: another flush cannot repair it. Callers
// that hand out references to objects (SetReference) flush first, so a ref
// can never point at an object that failed — or hasn't yet finished —
// uploading.
@@ -252,7 +253,5 @@ func (u *uploader) flush() error {
u.b.Flush()
u.mu.Lock()
defer u.mu.Unlock()
- err := u.err
- u.err = nil
- return err
+ return u.err
}
diff --git a/internal/storage/tigris/visibility_test.go b/internal/storage/tigris/visibility_test.go
new file mode 100644
index 0000000..da5a8ee
--- /dev/null
+++ b/internal/storage/tigris/visibility_test.go
@@ -0,0 +1,798 @@
+package tigris
+
+import (
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "errors"
+ "io"
+ "os/exec"
+ "slices"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "testing/synctest"
+ "time"
+
+ "github.com/aws/aws-sdk-go-v2/service/s3"
+ "github.com/go-git/go-git/v6/plumbing"
+)
+
+// visibilityClient separates successful uploads from read visibility. It is
+// deliberately not the default fake: supported buckets must preserve ordering.
+type visibilityClient struct {
+ s3API
+ hide atomic.Bool
+ mode string
+}
+
+func (c *visibilityClient) ListObjectsV2(ctx context.Context, in *s3.ListObjectsV2Input, opts ...func(*s3.Options)) (*s3.ListObjectsV2Output, error) {
+ out, err := c.s3API.ListObjectsV2(ctx, in, opts...)
+ if err == nil && c.hide.Load() && c.mode == "list" {
+ entries := out.Contents[:0]
+ for _, entry := range out.Contents {
+ if !strings.HasSuffix(sv(entry.Key), cueSuffix) {
+ entries = append(entries, entry)
+ }
+ }
+ out.Contents = entries
+ }
+ return out, err
+}
+
+func (c *visibilityClient) GetObject(ctx context.Context, in *s3.GetObjectInput, opts ...func(*s3.Options)) (*s3.GetObjectOutput, error) {
+ if c.hide.Load() && strings.HasSuffix(sv(in.Key), "."+c.mode) {
+ return nil, missingKeyErr()
+ }
+ return c.s3API.GetObject(ctx, in, opts...)
+}
+
+func TestPackVisibilityBoundaries(t *testing.T) {
+ if _, err := exec.LookPath("git"); err != nil {
+ t.Skip("git not installed")
+ }
+ fx := buildPackFixture(t, 3, false)
+ for _, tt := range []struct{ name string }{{"list"}, {"cue"}, {"bin"}} {
+ t.Run(tt.name, func(t *testing.T) {
+ f := newFakeS3(t)
+ writer := packedTestStorer(t, f)
+ writePack(t, writer, fx)
+ ref := plumbing.NewHashReference(plumbing.NewBranchReferenceName("main"), fx.hashes[0])
+ if err := writer.SetReference(ref); err != nil {
+ t.Fatal(err)
+ }
+ client := &visibilityClient{s3API: f, mode: tt.name}
+ client.hide.Store(true)
+ newReader := func() *Storer {
+ t.Helper()
+ s, err := New(context.Background(), "test-bucket", withClient(client), WithPackedRefs(true))
+ if err != nil {
+ t.Fatal(err)
+ }
+ return s
+ }
+ read := func(s *Storer) error {
+ obj, err := s.EncodedObject(plumbing.AnyObject, ref.Hash())
+ if err != nil {
+ return err
+ }
+ r, err := obj.Reader()
+ if err != nil {
+ return err
+ }
+ defer r.Close()
+ _, err = io.Copy(io.Discard, r)
+ return err
+ }
+ reader := newReader()
+ if got, err := reader.Reference(ref.Name()); err != nil || got.Hash() != ref.Hash() {
+ t.Fatalf("ref not visible: %v", err)
+ }
+ if err := read(reader); err == nil {
+ t.Fatal("hidden pack unexpectedly readable")
+ }
+ client.hide.Store(false)
+ if tt.name == "list" {
+ // A successful incomplete listing is cached, unlike a failed cue GET.
+ if err := read(reader); !errors.Is(err, plumbing.ErrObjectNotFound) {
+ t.Fatalf("cached incomplete index: %v", err)
+ }
+ }
+ if err := read(newReader()); err != nil {
+ t.Fatalf("fresh reader after visibility restored: %v", err)
+ }
+ })
+ }
+}
+
+// A reader that built its index before another writer published a ref must
+// recover without a new Storer. Misses within the cooldown share a refresh.
+func TestPackLookupRefreshOnMiss(t *testing.T) {
+ if _, err := exec.LookPath("git"); err != nil {
+ t.Skip("git not installed")
+ }
+ fx := buildPackFixture(t, 3, false)
+ for _, tt := range []struct{ name string }{{"object"}, {"has"}, {"size"}} {
+ t.Run(tt.name, func(t *testing.T) {
+ f := newFakeS3(t)
+ obs, snapshot := countingObserver()
+ reader := newTestStorer(t, f, obs)
+ if err := reader.ensurePacksBuilt(); err != nil {
+ t.Fatal(err)
+ }
+ writer := newTestStorer(t, f)
+ writePack(t, writer, fx)
+ if err := writer.up.flush(); err != nil {
+ t.Fatal(err)
+ }
+ h := fx.hashes[0]
+ switch tt.name {
+ case "object":
+ if _, err := reader.EncodedObject(plumbing.AnyObject, h); err != nil {
+ t.Fatal(err)
+ }
+ case "has":
+ if err := reader.HasEncodedObject(h); err != nil {
+ t.Fatal(err)
+ }
+ case "size":
+ if n, err := reader.EncodedObjectSize(h); err != nil || n != fx.byHash[h].size {
+ t.Fatalf("size=%d err=%v", n, err)
+ }
+ }
+ if n := snapshot()["ListObjectsV2"]; n != 2 {
+ t.Fatalf("list calls=%d, want 2", n)
+ }
+ ghost := hashForBody(reader.of, plumbing.BlobObject, "missing object")
+ for range 3 {
+ if err := reader.HasEncodedObject(ghost); !errors.Is(err, plumbing.ErrObjectNotFound) {
+ t.Fatal(err)
+ }
+ }
+ if n := snapshot()["ListObjectsV2"]; n != 2 {
+ t.Fatalf("repeated misses added listings: %d", n)
+ }
+ })
+ }
+}
+
+func TestPackRefreshCoalescesAndPreservesTables(t *testing.T) {
+ if _, err := exec.LookPath("git"); err != nil {
+ t.Skip("git not installed")
+ }
+ f := newFakeS3(t)
+ old := buildPackFixture(t, 2, false)
+ writer := newTestStorer(t, f)
+ writePack(t, writer, old)
+ if err := writer.up.flush(); err != nil {
+ t.Fatal(err)
+ }
+ obs, snapshot := countingObserver()
+ reader := newTestStorer(t, f, obs)
+ if err := reader.ensurePacksBuilt(); err != nil {
+ t.Fatal(err)
+ }
+ before := reader.packs.snapshot()
+ localHash := hashForBody(reader.of, plumbing.BlobObject, "local staged object")
+ reader.packs.register("local-staged", []cueRecord{{hash: localHash, typ: plumbing.BlobObject, raw: 19, stored: 19}}, "unused-local-path")
+ added := buildPackFixture(t, 4, false)
+ writePack(t, writer, added)
+ if err := writer.up.flush(); err != nil {
+ t.Fatal(err)
+ }
+ var target plumbing.Hash
+ for _, h := range added.hashes {
+ if _, _, found := before.lookup(h.Bytes()); !found {
+ target = h
+ break
+ }
+ }
+ if target.IsZero() {
+ t.Fatal("fixture added no new objects")
+ }
+ errs := make(chan error, 12)
+ for range 12 {
+ go func() { errs <- reader.HasEncodedObject(target) }()
+ }
+ for range 12 {
+ if err := <-errs; err != nil {
+ t.Fatal(err)
+ }
+ }
+ if n := snapshot()["ListObjectsV2"]; n != 2 {
+ t.Fatalf("list calls=%d want 2", n)
+ }
+ if err := reader.HasEncodedObject(localHash); err != nil {
+ t.Fatalf("lost local registration: %v", err)
+ }
+ for _, h := range old.hashes {
+ if err := reader.HasEncodedObject(h); err != nil {
+ t.Fatalf("lost old object: %v", err)
+ }
+ }
+ if _, _, found := before.lookup(target.Bytes()); found {
+ t.Fatal("refresh mutated old snapshot")
+ }
+}
+
+// A refresh runs only after loose storage has already said 404. A refresh that
+// then fails must not turn that absence into a storage error: go-git's
+// revision walk tolerates only ErrObjectNotFound for a `have` the server lacks.
+// Cancellation is the one exception, and a loose-storage error never reaches
+// the refresh at all.
+func TestPackRefreshErrors(t *testing.T) {
+ storageErr := errors.New("storage unavailable")
+ for _, tt := range []struct {
+ name string
+ want error
+ listings int // ListObjectsV2 calls, including the cold build
+ }{
+ {"permission", storageErr, 1},
+ {"list failure", plumbing.ErrObjectNotFound, 2},
+ {"cancelled", context.Canceled, 1},
+ {"corrupt cue", plumbing.ErrObjectNotFound, 2},
+ } {
+ t.Run(tt.name, func(t *testing.T) {
+ f := newFakeS3(t)
+ obs, snapshot := countingObserver()
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ reader, err := New(ctx, "test-bucket", withClient(f), obs)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := reader.ensurePacksBuilt(); err != nil {
+ t.Fatal(err)
+ }
+ switch tt.name {
+ case "permission":
+ f.getErr = storageErr
+ case "list failure":
+ f.listErr = storageErr
+ case "cancelled":
+ cancel()
+ case "corrupt cue":
+ f.objs["packs/corrupt.cue"] = fakeObject{body: []byte("invalid cue")}
+ }
+ h := hashForBody(reader.of, plumbing.BlobObject, "absent")
+ _, err = reader.EncodedObject(plumbing.AnyObject, h)
+ if !errors.Is(err, tt.want) {
+ t.Fatalf("error=%v want=%v", err, tt.want)
+ }
+ if n := snapshot()["ListObjectsV2"]; n != tt.listings {
+ t.Fatalf("list calls=%d, want %d", n, tt.listings)
+ }
+ })
+ }
+}
+
+type blockedPackList struct {
+ s3API
+ entered chan struct{}
+ release chan struct{}
+}
+
+func (c *blockedPackList) ListObjectsV2(ctx context.Context, in *s3.ListObjectsV2Input, opts ...func(*s3.Options)) (*s3.ListObjectsV2Output, error) {
+ close(c.entered)
+ select {
+ case <-c.release:
+ case <-ctx.Done():
+ return nil, ctx.Err()
+ }
+ return c.s3API.ListObjectsV2(ctx, in, opts...)
+}
+
+func TestPackRefreshDoesNotBlockKnownObjects(t *testing.T) {
+ if _, err := exec.LookPath("git"); err != nil {
+ t.Skip("git not installed")
+ }
+ f := newFakeS3(t)
+ fx := buildPackFixture(t, 2, false)
+ writer := newTestStorer(t, f)
+ writePack(t, writer, fx)
+ if err := writer.up.flush(); err != nil {
+ t.Fatal(err)
+ }
+ reader := newTestStorer(t, f)
+ if err := reader.ensurePacksBuilt(); err != nil {
+ t.Fatal(err)
+ }
+ gate := &blockedPackList{s3API: f, entered: make(chan struct{}), release: make(chan struct{})}
+ reader.client = gate
+ ghost := hashForBody(reader.of, plumbing.BlobObject, "absent")
+ missing := make(chan error, 1)
+ known := make(chan error, 1)
+ go func() { missing <- reader.HasEncodedObject(ghost) }()
+ defer func() {
+ close(gate.release)
+ select {
+ case <-missing:
+ case <-time.After(5 * time.Second):
+ t.Error("refresh stuck")
+ }
+ }()
+ waitPublication(t, gate.entered)
+ go func() { known <- reader.HasEncodedObject(fx.hashes[0]) }()
+ select {
+ case err := <-known:
+ if err != nil {
+ t.Fatal(err)
+ }
+ case <-time.After(time.Second):
+ t.Error("known object blocked behind pack refresh")
+ }
+}
+
+// fakeClock replaces a packIndex's clock, so a test moves through cooldowns
+// without sleeping. Advance only between calls on the same goroutine, or after
+// every goroutine that could read it has finished.
+type fakeClock struct{ t time.Time }
+
+func (c *fakeClock) now() time.Time { return c.t }
+
+func (c *fakeClock) advance(d time.Duration) { c.t = c.t.Add(d) }
+
+func installFakeClock(s *Storer) *fakeClock {
+ c := &fakeClock{t: time.Unix(1_700_000_000, 0)}
+ s.packs.now = c.now
+ return c
+}
+
+// syntheticPack builds one raw-codec blob pack without a git client. The
+// caller decides when its two keys land in the fake bucket.
+type syntheticPack struct {
+ id string
+ hash plumbing.Hash
+ recs []cueRecord
+ bin string
+ cue string
+}
+
+func makeSyntheticPack(s *Storer, body string) syntheticPack {
+ sum := sha256.Sum256([]byte(body))
+ recs := []cueRecord{{
+ hash: hashForBody(s.of, plumbing.BlobObject, body),
+ typ: plumbing.BlobObject,
+ codec: codecRaw,
+ stored: int64(len(body)),
+ raw: int64(len(body)),
+ }}
+ return syntheticPack{
+ id: hex.EncodeToString(sum[:]),
+ hash: recs[0].hash,
+ recs: recs,
+ bin: body,
+ cue: string(encodeCue(s.oh.Size(), recs)),
+ }
+}
+
+func (p syntheticPack) put(s *Storer, f *fakeS3) {
+ f.put(s.prefix+packPrefix+p.id+binSuffix, p.bin, nil)
+ f.put(s.prefix+packPrefix+p.id+cueSuffix, p.cue, nil)
+}
+
+// A missing probe or a transient error must not poison a long-lived reader.
+func TestPackRefreshRecoversAfterCooldown(t *testing.T) {
+ if _, err := exec.LookPath("git"); err != nil {
+ t.Skip("git not installed")
+ }
+ fixtures := []packFixture{buildPackFixture(t, 2, false), buildPackFixture(t, 5, false)}
+ for _, tt := range []struct {
+ name string
+ fail bool
+ }{{"missing probe", false}, {"listing failure", true}} {
+ t.Run(tt.name, func(t *testing.T) {
+ f := newFakeS3(t)
+ obs, counts := countingObserver()
+ reader := newTestStorer(t, f, obs)
+ clock := installFakeClock(reader)
+ if err := reader.ensurePacksBuilt(); err != nil {
+ t.Fatal(err)
+ }
+ ghost := hashForBody(reader.of, plumbing.BlobObject, "absent")
+ if tt.fail {
+ f.listErr = errors.New("temporary listing failure")
+ }
+ for range 12 {
+ if err := reader.HasEncodedObject(ghost); !errors.Is(err, plumbing.ErrObjectNotFound) {
+ t.Fatalf("initial miss: %v", err)
+ }
+ }
+ if n := counts()["ListObjectsV2"]; n != 2 {
+ t.Fatalf("miss burst made %d listings, want 2", n)
+ }
+ f.listErr = nil
+ writer := newTestStorer(t, f)
+ for round, fx := range fixtures {
+ before := reader.packs.snapshot()
+ writePack(t, writer, fx)
+ if err := writer.up.flush(); err != nil {
+ t.Fatal(err)
+ }
+ var target plumbing.Hash
+ for _, h := range fx.hashes {
+ if _, _, found := before.lookup(h.Bytes()); !found {
+ target = h
+ break
+ }
+ }
+ if target.IsZero() {
+ t.Fatal("fixture has no new objects")
+ }
+ // The first cooldown, and the one after a refresh that found a
+ // pack, are both the base length.
+ clock.advance(time.Second)
+ errs := make(chan error, 12)
+ for i := range 12 {
+ go func() {
+ var err error
+ switch i % 3 {
+ case 0:
+ err = reader.HasEncodedObject(target)
+ case 1:
+ _, err = reader.EncodedObjectSize(target)
+ case 2:
+ _, err = reader.EncodedObject(plumbing.AnyObject, target)
+ }
+ errs <- err
+ }()
+ }
+ for range 12 {
+ if err := <-errs; err != nil {
+ t.Fatalf("publication %d: %v", round+1, err)
+ }
+ }
+ if n := counts()["ListObjectsV2"]; n != round+3 {
+ t.Fatalf("publication %d: %d listings, want %d", round+1, n, round+3)
+ }
+ }
+ })
+ }
+}
+
+// A slow listing must not let queued misses start another attempt immediately.
+func TestPackRefreshCooldownStartsAfterCompletion(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ f := newFakeS3(t)
+ obs, counts := countingObserver()
+ reader := newTestStorer(t, f, obs)
+ if err := reader.ensurePacksBuilt(); err != nil {
+ t.Fatal(err)
+ }
+ gate := &blockedPackList{s3API: f, entered: make(chan struct{}), release: make(chan struct{})}
+ reader.client = gate
+ errs := make(chan error, 12)
+ for range 12 {
+ go func() { errs <- reader.refreshPacks() }()
+ }
+ <-gate.entered
+ synctest.Wait()
+ time.Sleep(2 * time.Second)
+ close(gate.release)
+ for range 12 {
+ if err := <-errs; err != nil {
+ t.Fatal(err)
+ }
+ }
+ reader.client = f
+ if err := reader.refreshPacks(); err != nil {
+ t.Fatal(err)
+ }
+ if n := counts()["ListObjectsV2"]; n != 2 {
+ t.Fatalf("slow attempt caused %d listings, want 2", n)
+ }
+ time.Sleep(time.Second)
+ if err := reader.refreshPacks(); err != nil {
+ t.Fatal(err)
+ }
+ if n := counts()["ListObjectsV2"]; n != 3 {
+ t.Fatalf("after cooldown: %d listings, want 3", n)
+ }
+ })
+}
+
+// The cooldown adapts to what a refresh finds. An attempt that adds nothing,
+// or fails, doubles the next cooldown up to the cap. An attempt that adds a
+// pack resets it to the base. A reader is never told to stop refreshing: a
+// fresh reader fetching a just-pushed ref is the case this path exists for.
+func TestPackRefreshCooldownAdapts(t *testing.T) {
+ for _, tt := range []struct {
+ name string
+ fail bool
+ }{{"empty listings", false}, {"listing failures", true}} {
+ t.Run(tt.name, func(t *testing.T) {
+ f := newFakeS3(t)
+ obs, counts := countingObserver()
+ reader := newTestStorer(t, f, obs)
+ clock := installFakeClock(reader)
+ if err := reader.ensurePacksBuilt(); err != nil {
+ t.Fatal(err)
+ }
+ if tt.fail {
+ f.listErr = errors.New("temporary listing failure")
+ }
+ listings := 1 // the cold build
+ refresh := func() {
+ t.Helper()
+ err := reader.refreshPacks()
+ if (err != nil) != tt.fail {
+ t.Fatalf("refresh error=%v, want failure=%v", err, tt.fail)
+ }
+ }
+ for i, want := range []time.Duration{time.Second, 2 * time.Second, 4 * time.Second, 8 * time.Second, 10 * time.Second, 10 * time.Second} {
+ refresh()
+ listings++
+ if got := reader.packs.refreshWait; got != want {
+ t.Fatalf("attempt %d: cooldown=%v, want %v", i+1, got, want)
+ }
+ if n := counts()["ListObjectsV2"]; n != listings {
+ t.Fatalf("attempt %d: %d listings, want %d", i+1, n, listings)
+ }
+ // A miss just inside the cooldown lists nothing.
+ clock.advance(want - time.Millisecond)
+ if err := reader.refreshPacks(); err != nil {
+ t.Fatalf("attempt %d: cooldown returned %v", i+1, err)
+ }
+ if n := counts()["ListObjectsV2"]; n != listings {
+ t.Fatalf("attempt %d: miss inside the cooldown listed (%d listings)", i+1, n)
+ }
+ clock.advance(time.Millisecond)
+ }
+
+ f.listErr = nil
+ pack := makeSyntheticPack(reader, "published while the reader backed off")
+ pack.put(reader, f)
+ if err := reader.refreshPacks(); err != nil {
+ t.Fatal(err)
+ }
+ if got := reader.packs.refreshWait; got != time.Second {
+ t.Fatalf("a productive refresh left the cooldown at %v, want %v", got, time.Second)
+ }
+ if err := reader.HasEncodedObject(pack.hash); err != nil {
+ t.Fatalf("new pack not visible: %v", err)
+ }
+ })
+ }
+}
+
+// blockedCueGet holds the first .cue GetObject open until release closes.
+type blockedCueGet struct {
+ s3API
+ entered chan struct{}
+ release chan struct{}
+ once sync.Once
+}
+
+func (c *blockedCueGet) GetObject(ctx context.Context, in *s3.GetObjectInput, opts ...func(*s3.Options)) (*s3.GetObjectOutput, error) {
+ if strings.HasSuffix(sv(in.Key), cueSuffix) {
+ c.once.Do(func() { close(c.entered) })
+ select {
+ case <-c.release:
+ case <-ctx.Done():
+ return nil, ctx.Err()
+ }
+ }
+ return c.s3API.GetObject(ctx, in, opts...)
+}
+
+// tablesHolding counts the tables that carry records of pack num.
+func tablesHolding(p *packIndex, num uint32) int {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ n := 0
+ for _, t := range p.tables {
+ if slices.ContainsFunc(t.recs, func(r tableRec) bool { return r.pack == num }) {
+ n++
+ }
+ }
+ return n
+}
+
+// A pack this storer registers while a refresh is in flight must not be loaded
+// a second time from its own freshly landed .cue. The write can race the
+// listing, or the cue fetch that follows it.
+func TestPackRefreshSkipsPacksRegisteredMeanwhile(t *testing.T) {
+ if _, err := exec.LookPath("git"); err != nil {
+ t.Skip("git not installed")
+ }
+ fx := buildPackFixture(t, 3, false)
+ for _, tt := range []struct {
+ name string
+ // published is true when another writer lands the same pack before
+ // the refresh lists, so the listing sees it and the race is on the
+ // cue fetch instead of the listing.
+ published bool
+ }{{"during the listing", false}, {"during the cue fetch", true}} {
+ t.Run(tt.name, func(t *testing.T) {
+ f := newFakeS3(t)
+ reader := newTestStorer(t, f)
+ if err := reader.ensurePacksBuilt(); err != nil {
+ t.Fatal(err)
+ }
+ var entered, release chan struct{}
+ if tt.published {
+ other := newTestStorer(t, f)
+ writePack(t, other, fx)
+ if err := other.up.flush(); err != nil {
+ t.Fatal(err)
+ }
+ gate := &blockedCueGet{s3API: f, entered: make(chan struct{}), release: make(chan struct{})}
+ entered, release = gate.entered, gate.release
+ reader.client = gate
+ } else {
+ gate := &blockedPackList{s3API: f, entered: make(chan struct{}), release: make(chan struct{})}
+ entered, release = gate.entered, gate.release
+ reader.client = gate
+ }
+
+ ghost := hashForBody(reader.of, plumbing.BlobObject, "absent")
+ missing := make(chan error, 1)
+ go func() { missing <- reader.HasEncodedObject(ghost) }()
+ waitPublication(t, entered)
+
+ // The normal write path: register, then upload the bin and cue.
+ writePack(t, reader, fx)
+ if err := reader.up.flush(); err != nil {
+ t.Fatal(err)
+ }
+ close(release)
+ select {
+ case err := <-missing:
+ if !errors.Is(err, plumbing.ErrObjectNotFound) {
+ t.Fatalf("miss: %v", err)
+ }
+ case <-time.After(10 * time.Second):
+ t.Fatal("refresh stuck")
+ }
+
+ f.mu.Lock()
+ var ids []string
+ for k := range f.objs {
+ if strings.HasSuffix(k, cueSuffix) {
+ ids = append(ids, strings.TrimSuffix(strings.TrimPrefix(k, packPrefix), cueSuffix))
+ }
+ }
+ f.mu.Unlock()
+ if len(ids) != 1 {
+ t.Fatalf("bucket holds %d packs, want 1 (same content, same id)", len(ids))
+ }
+ reader.packs.mu.Lock()
+ num, ok := reader.packs.idNum[ids[0]]
+ reader.packs.mu.Unlock()
+ if !ok {
+ t.Fatalf("pack %s is not numbered", ids[0])
+ }
+ if n := tablesHolding(reader.packs, num); n != 1 {
+ t.Fatalf("pack %s is in %d tables, want 1", ids[0], n)
+ }
+ for _, h := range fx.hashes {
+ if err := reader.HasEncodedObject(h); err != nil {
+ t.Fatalf("own pack not visible: %v", err)
+ }
+ }
+ })
+ }
+}
+
+// deregister runs when an upload reports failure, but the bin and cue can be
+// durable regardless: the PUT succeeded server-side and only the reply was
+// lost. A later refresh on the same storer must then load the pack from the
+// bucket like any other, including when its table was shared with other packs
+// and deregister could only mark it dead.
+func TestPackRefreshReloadsDeregisteredPack(t *testing.T) {
+ for _, tt := range []struct {
+ name string
+ shared bool
+ }{{"own table", false}, {"shared table", true}} {
+ t.Run(tt.name, func(t *testing.T) {
+ f := newFakeS3(t)
+ reader := newTestStorer(t, f)
+ clock := installFakeClock(reader)
+ if err := reader.ensurePacksBuilt(); err != nil {
+ t.Fatal(err)
+ }
+ pack := makeSyntheticPack(reader, "durable despite a failed upload reply")
+ reader.packs.register(pack.id, pack.recs, "unused-local-path")
+ if tt.shared {
+ other := makeSyntheticPack(reader, "a neighbour in the same table")
+ reader.packs.register(other.id, other.recs, "unused-local-path")
+ reader.packs.mu.Lock()
+ reader.packs.tables = []*packTable{mergeTables(reader.packs.hashSize, reader.packs.tables, nil)}
+ reader.packs.mu.Unlock()
+ }
+ reader.packs.deregister(pack.id)
+ if err := reader.HasEncodedObject(pack.hash); !errors.Is(err, plumbing.ErrObjectNotFound) {
+ t.Fatalf("deregistered pack still visible: %v", err)
+ }
+
+ pack.put(reader, f)
+ clock.advance(reader.packs.refreshWait)
+ if err := reader.HasEncodedObject(pack.hash); err != nil {
+ t.Fatalf("durable pack not reloaded: %v", err)
+ }
+ obj, err := reader.EncodedObject(plumbing.BlobObject, pack.hash)
+ if err != nil {
+ t.Fatal(err)
+ }
+ r, err := obj.Reader()
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer r.Close()
+ body, err := io.ReadAll(r)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if string(body) != pack.bin {
+ t.Fatalf("body=%q want %q", body, pack.bin)
+ }
+ })
+ }
+}
+
+// Model a storage call that cannot stop promptly on context cancellation.
+type uncancelablePackList struct{ *blockedPackList }
+
+func (c *uncancelablePackList) ListObjectsV2(ctx context.Context, in *s3.ListObjectsV2Input, opts ...func(*s3.Options)) (*s3.ListObjectsV2Output, error) {
+ return c.blockedPackList.ListObjectsV2(context.WithoutCancel(ctx), in, opts...)
+}
+
+// A canceled storer reports the cancellation from refreshPacks whether the
+// cancel lands during a cooldown or while a caller waits on another caller's
+// slow listing.
+func TestPackRefreshCancellation(t *testing.T) {
+ for _, tt := range []struct {
+ name string
+ // waiting is true when the cancel lands while a second caller waits
+ // on a listing that itself ignores cancellation.
+ waiting bool
+ }{{"during the cooldown", false}, {"while waiting on a slow listing", true}} {
+ t.Run(tt.name, func(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ f := newFakeS3(t)
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ reader, err := New(ctx, "test-bucket", withClient(f))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := reader.ensurePacksBuilt(); err != nil {
+ t.Fatal(err)
+ }
+
+ if !tt.waiting {
+ if err := reader.refreshPacks(); err != nil {
+ t.Fatal(err)
+ }
+ cancel()
+ if err := reader.refreshPacks(); !errors.Is(err, context.Canceled) {
+ t.Fatalf("cooldown masked cancellation: %v", err)
+ }
+ return
+ }
+
+ gate := &blockedPackList{s3API: f, entered: make(chan struct{}), release: make(chan struct{})}
+ reader.client = &uncancelablePackList{gate}
+ leader := make(chan error, 1)
+ waiter := make(chan error, 1)
+ go func() { leader <- reader.refreshPacks() }()
+ <-gate.entered
+ go func() { waiter <- reader.refreshPacks() }()
+ synctest.Wait()
+ cancel()
+ synctest.Wait()
+ select {
+ case err := <-waiter:
+ if !errors.Is(err, context.Canceled) {
+ t.Errorf("waiter error=%v", err)
+ }
+ default:
+ t.Error("canceled waiter remained blocked on storage")
+ }
+ close(gate.release)
+ <-leader
+ })
+ })
+ }
+}