Git backed by object storage because you can't stop me
git
object-storage
kefka
refactor(protocol): rename daemon files to git_protocol
Move daemon.go and daemon_test.go to git_protocol.go and git_protocol_test.go
to better reflect their purpose. No logic changes.
Signed-off-by: Xe Iaso <me@xeiaso.net>
+870
-30
+17
-1
cmd/objgitd/daemon.go
cmd/objgitd/git_protocol.go
+17
-1
cmd/objgitd/daemon.go
cmd/objgitd/git_protocol.go
···
26
26
// It is cleared once the (possibly long) transfer begins.
27
27
const handshakeTimeout = 30 * time.Second
28
28
29
+
// streamingStorer wraps a storage.Storer to hide its optional
30
+
// storer.PackfileWriter capability. go-git's UpdateObjectStorage drains the
31
+
// incoming pack into PackfileWriter via io.CopyBuffer, which only returns on
32
+
// io.EOF — fine over HTTP (the request body has a natural EOF) but a deadlock
33
+
// over git://, where the client holds the connection open waiting for the
34
+
// server's report-status. With PackfileWriter hidden, UpdateObjectStorage
35
+
// falls through to Parser.Parse, which knows the end of the pack from the
36
+
// pack format itself and never waits for an EOF.
37
+
//
38
+
// Trade-off: Parser.Parse writes loose objects (one Rename → one S3 PUT each)
39
+
// instead of one packfile, so large git:// pushes incur more S3 calls. HTTP
40
+
// keeps the fast PackfileWriter path.
41
+
type streamingStorer struct {
42
+
storage.Storer
43
+
}
44
+
29
45
// daemon serves the git:// (TCP) protocol out of a billy filesystem.
30
46
type daemon struct {
31
47
fs billy.Filesystem
···
119
135
_, _ = pktline.WriteError(conn, fmt.Errorf("cannot open repository %q", req.Pathname))
120
136
return fmt.Errorf("opening %q for push: %w", req.Pathname, err)
121
137
}
122
-
return transport.ReceivePack(ctx, st, r, conn, &transport.ReceivePackRequest{
138
+
return transport.ReceivePack(ctx, streamingStorer{Storer: st}, r, conn, &transport.ReceivePackRequest{
123
139
GitProtocol: gitProtocol,
124
140
})
125
141
cmd/objgitd/daemon_test.go
cmd/objgitd/git_protocol_test.go
cmd/objgitd/daemon_test.go
cmd/objgitd/git_protocol_test.go
+169
cmd/objgitd/http.go
+169
cmd/objgitd/http.go
···
1
+
package main
2
+
3
+
import (
4
+
"compress/gzip"
5
+
"errors"
6
+
"fmt"
7
+
"io"
8
+
"log/slog"
9
+
"net/http"
10
+
"net/url"
11
+
"strings"
12
+
13
+
"github.com/go-git/go-git/v6/plumbing/transport"
14
+
"github.com/go-git/go-git/v6/storage"
15
+
"github.com/go-git/go-git/v6/utils/ioutil"
16
+
)
17
+
18
+
// ServeHTTP speaks the git smart-HTTP protocol. It dispatches on the URL suffix
19
+
// the way git-http-backend does: repository paths are variable-depth (e.g.
20
+
// /foo/bar.git) and precede a fixed endpoint suffix, which http.ServeMux's
21
+
// wildcards cannot express.
22
+
func (d *daemon) ServeHTTP(w http.ResponseWriter, r *http.Request) {
23
+
p := r.URL.Path
24
+
switch {
25
+
case r.Method == http.MethodGet && strings.HasSuffix(p, "/info/refs"):
26
+
d.handleInfoRefs(w, r, strings.TrimSuffix(p, "/info/refs"))
27
+
case r.Method == http.MethodPost && strings.HasSuffix(p, "/git-upload-pack"):
28
+
d.handleRPC(w, r, transport.UploadPackService, strings.TrimSuffix(p, "/git-upload-pack"))
29
+
case r.Method == http.MethodPost && strings.HasSuffix(p, "/git-receive-pack"):
30
+
d.handleRPC(w, r, transport.ReceivePackService, strings.TrimSuffix(p, "/git-receive-pack"))
31
+
default:
32
+
http.NotFound(w, r)
33
+
}
34
+
}
35
+
36
+
// handleInfoRefs serves the reference-discovery phase:
37
+
// GET /{repo}/info/refs?service=git-(upload|receive)-pack.
38
+
func (d *daemon) handleInfoRefs(w http.ResponseWriter, r *http.Request, repoPath string) {
39
+
service := r.URL.Query().Get("service")
40
+
switch service {
41
+
case transport.UploadPackService, transport.ReceivePackService:
42
+
default:
43
+
http.Error(w, fmt.Sprintf("unsupported service %q", service), http.StatusBadRequest)
44
+
return
45
+
}
46
+
47
+
st, ok := d.resolve(w, service, repoPath)
48
+
if !ok {
49
+
return
50
+
}
51
+
52
+
slog.Info("serving smart-http advertisement",
53
+
"service", service,
54
+
"path", repoPath,
55
+
"remote", r.RemoteAddr,
56
+
)
57
+
58
+
w.Header().Set("Content-Type", "application/x-"+service+"-advertisement")
59
+
w.Header().Set("Cache-Control", "no-cache")
60
+
61
+
gitProtocol := r.Header.Get("Git-Protocol")
62
+
out := ioutil.WriteNopCloser(w)
63
+
64
+
// AdvertiseRefs+StatelessRPC emits the "# service=...\n" smart-reply prefix
65
+
// followed by the ref advertisement, then returns without touching a reader.
66
+
var err error
67
+
switch service {
68
+
case transport.UploadPackService:
69
+
err = transport.UploadPack(r.Context(), st, nil, out, &transport.UploadPackRequest{
70
+
AdvertiseRefs: true,
71
+
StatelessRPC: true,
72
+
GitProtocol: gitProtocol,
73
+
})
74
+
case transport.ReceivePackService:
75
+
err = transport.ReceivePack(r.Context(), st, nil, out, &transport.ReceivePackRequest{
76
+
AdvertiseRefs: true,
77
+
StatelessRPC: true,
78
+
GitProtocol: gitProtocol,
79
+
})
80
+
}
81
+
if err != nil {
82
+
slog.Error("smart-http advertisement failed", "service", service, "path", repoPath, "err", err)
83
+
}
84
+
}
85
+
86
+
// handleRPC serves a stateless negotiation round:
87
+
// POST /{repo}/git-(upload|receive)-pack.
88
+
func (d *daemon) handleRPC(w http.ResponseWriter, r *http.Request, service, repoPath string) {
89
+
st, ok := d.resolve(w, service, repoPath)
90
+
if !ok {
91
+
return
92
+
}
93
+
94
+
body := r.Body
95
+
if r.Header.Get("Content-Encoding") == "gzip" {
96
+
gz, err := gzip.NewReader(r.Body)
97
+
if err != nil {
98
+
http.Error(w, "invalid gzip body", http.StatusBadRequest)
99
+
return
100
+
}
101
+
defer gz.Close()
102
+
body = gz
103
+
}
104
+
105
+
slog.Info("serving smart-http rpc",
106
+
"service", service,
107
+
"path", repoPath,
108
+
"remote", r.RemoteAddr,
109
+
)
110
+
111
+
w.Header().Set("Content-Type", "application/x-"+service+"-result")
112
+
w.Header().Set("Cache-Control", "no-cache")
113
+
114
+
// The server commands call Close between negotiation steps; the body and the
115
+
// response writer must survive that, so both are wrapped as no-op closers.
116
+
in := io.NopCloser(body)
117
+
out := ioutil.WriteNopCloser(w)
118
+
gitProtocol := r.Header.Get("Git-Protocol")
119
+
120
+
var err error
121
+
switch service {
122
+
case transport.UploadPackService:
123
+
err = transport.UploadPack(r.Context(), st, in, out, &transport.UploadPackRequest{
124
+
StatelessRPC: true,
125
+
GitProtocol: gitProtocol,
126
+
})
127
+
case transport.ReceivePackService:
128
+
err = transport.ReceivePack(r.Context(), st, in, out, &transport.ReceivePackRequest{
129
+
StatelessRPC: true,
130
+
GitProtocol: gitProtocol,
131
+
})
132
+
}
133
+
if err != nil {
134
+
// The status line is already sent, so this can only be logged.
135
+
slog.Error("smart-http rpc failed", "service", service, "path", repoPath, "err", err)
136
+
}
137
+
}
138
+
139
+
// resolve loads the storer for an HTTP request, applying the same rules as the
140
+
// git:// handler: anonymous read, push gated by allowPush, and create-on-first-
141
+
// push. It writes an HTTP error and returns ok=false when the request cannot
142
+
// proceed.
143
+
func (d *daemon) resolve(w http.ResponseWriter, service, repoPath string) (storage.Storer, bool) {
144
+
if service == transport.ReceivePackService {
145
+
if !d.allowPush {
146
+
http.Error(w, "push is disabled on this server", http.StatusForbidden)
147
+
return nil, false
148
+
}
149
+
st, err := d.loadOrInit(repoPath)
150
+
if err != nil {
151
+
slog.Error("opening repository for push", "path", repoPath, "err", err)
152
+
http.Error(w, "cannot open repository", http.StatusInternalServerError)
153
+
return nil, false
154
+
}
155
+
return st, true
156
+
}
157
+
158
+
st, err := d.loader.Load(&url.URL{Path: repoPath})
159
+
if err != nil {
160
+
if errors.Is(err, transport.ErrRepositoryNotFound) {
161
+
http.Error(w, "repository not found", http.StatusNotFound)
162
+
return nil, false
163
+
}
164
+
slog.Error("loading repository", "path", repoPath, "err", err)
165
+
http.Error(w, "cannot open repository", http.StatusInternalServerError)
166
+
return nil, false
167
+
}
168
+
return st, true
169
+
}
+124
cmd/objgitd/http_test.go
+124
cmd/objgitd/http_test.go
···
1
+
package main
2
+
3
+
import (
4
+
"net/http/httptest"
5
+
"os/exec"
6
+
"path/filepath"
7
+
"strings"
8
+
"testing"
9
+
10
+
"github.com/go-git/go-billy/v6"
11
+
"github.com/go-git/go-billy/v6/memfs"
12
+
"github.com/go-git/go-git/v6/plumbing/transport"
13
+
)
14
+
15
+
// TestSmartHTTP drives a real git client against the smart-HTTP handler over an
16
+
// in-memory filesystem, covering push (create-on-demand), the allowPush gate,
17
+
// and clone round-trips.
18
+
func TestSmartHTTP(t *testing.T) {
19
+
if _, err := exec.LookPath("git"); err != nil {
20
+
t.Skip("git not installed")
21
+
}
22
+
23
+
for _, tt := range []struct {
24
+
name string
25
+
allowPush bool
26
+
doPush bool
27
+
wantPushErr bool
28
+
wantCloneErr bool
29
+
}{
30
+
{
31
+
name: "push creates repo and clone round-trips",
32
+
allowPush: true,
33
+
doPush: true,
34
+
},
35
+
{
36
+
name: "push rejected when disabled",
37
+
allowPush: false,
38
+
doPush: true,
39
+
wantPushErr: true,
40
+
wantCloneErr: true,
41
+
},
42
+
{
43
+
name: "clone of missing repo fails",
44
+
allowPush: true,
45
+
doPush: false,
46
+
wantCloneErr: true,
47
+
},
48
+
} {
49
+
t.Run(tt.name, func(t *testing.T) {
50
+
ts, fs := newHTTPServer(t, tt.allowPush)
51
+
remote := ts.URL + "/test.git"
52
+
53
+
var srcHead string
54
+
if tt.doPush {
55
+
work := seedRepo(t)
56
+
srcHead = strings.TrimSpace(runGit(t, work, "rev-parse", "HEAD"))
57
+
58
+
out, err := tryGit(work, "push", remote, "main")
59
+
if tt.wantPushErr {
60
+
if err == nil {
61
+
t.Fatalf("expected push to be rejected, got success:\n%s", out)
62
+
}
63
+
} else if err != nil {
64
+
t.Fatalf("push failed: %v\n%s", err, out)
65
+
}
66
+
}
67
+
68
+
// The bare repo must exist on disk iff a push was expected to land.
69
+
_, statErr := fs.Stat("/test.git/config")
70
+
pushLanded := tt.doPush && !tt.wantPushErr
71
+
if pushLanded && statErr != nil {
72
+
t.Fatalf("expected repo to be created on push, but config missing: %v", statErr)
73
+
}
74
+
if !pushLanded && statErr == nil {
75
+
t.Fatal("repository must not exist when push did not land")
76
+
}
77
+
78
+
dst := t.TempDir()
79
+
out, err := tryGit(dst, "clone", remote, "cloned")
80
+
if tt.wantCloneErr {
81
+
if err == nil {
82
+
t.Fatalf("expected clone to fail, got success:\n%s", out)
83
+
}
84
+
return
85
+
}
86
+
if err != nil {
87
+
t.Fatalf("clone failed: %v\n%s", err, out)
88
+
}
89
+
90
+
gotHead := strings.TrimSpace(runGit(t, filepath.Join(dst, "cloned"), "rev-parse", "HEAD"))
91
+
if gotHead != srcHead {
92
+
t.Logf("want: %s", srcHead)
93
+
t.Logf("got: %s", gotHead)
94
+
t.Error("cloned HEAD does not match pushed HEAD")
95
+
}
96
+
})
97
+
}
98
+
}
99
+
100
+
// newHTTPServer starts an httptest server backed by a fresh in-memory filesystem
101
+
// and returns it alongside that filesystem for state assertions.
102
+
func newHTTPServer(t *testing.T, allowPush bool) (*httptest.Server, billy.Filesystem) {
103
+
t.Helper()
104
+
fs := memfs.New()
105
+
d := &daemon{
106
+
fs: fs,
107
+
loader: transport.NewFilesystemLoader(fs, false),
108
+
allowPush: allowPush,
109
+
}
110
+
ts := httptest.NewServer(d)
111
+
t.Cleanup(ts.Close)
112
+
return ts, fs
113
+
}
114
+
115
+
// seedRepo creates a local git repository with one commit and returns its path.
116
+
func seedRepo(t *testing.T) string {
117
+
t.Helper()
118
+
work := t.TempDir()
119
+
runGit(t, work, "init", "-b", "main")
120
+
runGit(t, work, "config", "user.email", "test@example.com")
121
+
runGit(t, work, "config", "user.name", "Test")
122
+
runGit(t, work, "commit", "--allow-empty", "-m", "initial")
123
+
return work
124
+
}
+52
-13
cmd/objgitd/main.go
+52
-13
cmd/objgitd/main.go
···
2
2
3
3
import (
4
4
"context"
5
+
"errors"
5
6
"flag"
7
+
"fmt"
6
8
"log/slog"
7
9
"net"
10
+
"net/http"
8
11
"os"
9
12
"os/signal"
10
13
"syscall"
14
+
"time"
11
15
12
16
"github.com/facebookgo/flagenv"
13
17
"github.com/go-git/go-git/v6/plumbing/transport"
14
18
"github.com/tigrisdata/storage-go"
19
+
"golang.org/x/sync/errgroup"
20
+
"tangled.org/xeiaso.net/objgit/internal"
15
21
"tangled.org/xeiaso.net/objgit/internal/s3fs"
16
22
17
23
_ "github.com/joho/godotenv/autoload"
18
24
)
19
25
20
26
var (
21
-
bind = flag.String("bind", ":9418", "TCP address to listen on for the git:// protocol")
27
+
gitBind = flag.String("git-bind", ":9418", "TCP address to listen on for the git:// protocol; empty disables it")
28
+
httpBind = flag.String("http-bind", ":8080", "TCP address to listen on for the git smart-HTTP protocol; empty disables it")
22
29
bucket = flag.String("bucket", "", "Tigris bucket that holds the git repositories")
23
30
allowPush = flag.Bool("allow-push", false, "allow unauthenticated git-receive-pack (push) requests")
24
31
slogLevel = flag.String("slog-level", "INFO", "log level (DEBUG, INFO, WARN, ERROR)")
···
28
35
flagenv.Parse()
29
36
flag.Parse()
30
37
31
-
var lvl slog.Level
32
-
if err := lvl.UnmarshalText([]byte(*slogLevel)); err != nil {
33
-
slog.Error("invalid -slog-level", "value", *slogLevel, "err", err)
38
+
logger, err := internal.InitSlog(*slogLevel)
39
+
if err != nil {
40
+
fmt.Fprintln(os.Stderr, "error initializing logging stack:", err)
34
41
os.Exit(1)
35
42
}
36
-
slog.SetDefault(slog.New(slog.NewJSONHandler(os.Stderr, &slog.HandlerOptions{Level: lvl})))
43
+
slog.SetDefault(logger)
37
44
38
45
if *bucket == "" {
39
46
slog.Error("-bucket is required")
40
47
os.Exit(1)
41
48
}
42
49
50
+
if *gitBind == "" && *httpBind == "" {
51
+
slog.Error("at least one of -git-bind or -http-bind must be set")
52
+
os.Exit(1)
53
+
}
54
+
43
55
ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
44
56
defer cancel()
45
57
···
61
73
allowPush: *allowPush,
62
74
}
63
75
64
-
ln, err := net.Listen("tcp", *bind)
65
-
if err != nil {
66
-
slog.Error("can't listen", "bind", *bind, "err", err)
67
-
os.Exit(1)
68
-
}
69
-
70
76
slog.Info("objgitd listening",
71
-
"bind", *bind,
77
+
"git_bind", *gitBind,
78
+
"http_bind", *httpBind,
72
79
"bucket", *bucket,
73
80
"allow_push", *allowPush,
74
81
)
75
82
76
-
if err := d.Serve(ctx, ln); err != nil {
83
+
g, gCtx := errgroup.WithContext(ctx)
84
+
85
+
if *gitBind != "" {
86
+
ln, err := net.Listen("tcp", *gitBind)
87
+
if err != nil {
88
+
slog.Error("can't listen", "git_bind", *gitBind, "err", err)
89
+
os.Exit(1)
90
+
}
91
+
g.Go(func() error { return d.Serve(gCtx, ln) })
92
+
}
93
+
94
+
if *httpBind != "" {
95
+
ln, err := net.Listen("tcp", *httpBind)
96
+
if err != nil {
97
+
slog.Error("can't listen", "http_bind", *httpBind, "err", err)
98
+
os.Exit(1)
99
+
}
100
+
srv := &http.Server{Handler: d}
101
+
g.Go(func() error {
102
+
if err := srv.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) {
103
+
return err
104
+
}
105
+
return nil
106
+
})
107
+
g.Go(func() error {
108
+
<-gCtx.Done()
109
+
shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
110
+
defer cancel()
111
+
return srv.Shutdown(shutdownCtx)
112
+
})
113
+
}
114
+
115
+
if err := g.Wait(); err != nil {
77
116
slog.Error("server stopped", "err", err)
78
117
os.Exit(1)
79
118
}
+38
-7
internal/s3fs/basic.go
+38
-7
internal/s3fs/basic.go
···
3
3
package s3fs
4
4
5
5
import (
6
+
"bytes"
6
7
"context"
7
8
"errors"
8
9
"fmt"
···
10
11
"os"
11
12
"path"
12
13
"strings"
14
+
"time"
13
15
14
16
"github.com/aws/aws-sdk-go-v2/aws"
15
17
"github.com/aws/aws-sdk-go-v2/service/s3"
···
66
68
return newS3DirFile(key, fs3.bucket, fs3.client), nil
67
69
}
68
70
71
+
// A TempFile that has not yet been renamed lives only in memory; serve
72
+
// reads from that buffer so go-git's PackWriter can read the pack back
73
+
// while it is still being written.
74
+
if buf, ok := fs3.lookupTemp(filename); ok {
75
+
return &tempReadFile{buf: buf, name: filename}, nil
76
+
}
77
+
69
78
f, err := newS3ReadFile(fs3.client, fs3.bucket, key, filename)
70
79
if err == nil {
71
80
return f, nil
···
118
127
return newDirInfo("/"), nil
119
128
}
120
129
130
+
// A still-open TempFile lives only in memory; report its current size so
131
+
// callers that Stat the temp path before Rename see a consistent view.
132
+
if buf, ok := fs3.lookupTemp(filename); ok {
133
+
return newFileInfo(path.Base(filename), buf.size(), time.Now()), nil
134
+
}
135
+
121
136
ctx := context.TODO()
122
137
123
138
head, err := fs3.client.HeadObject(ctx, &s3.HeadObjectInput{
···
160
175
return nil, &os.PathError{Op: "stat", Path: filename, Err: fs.ErrNotExist}
161
176
}
162
177
163
-
// Rename renames (moves) oldpath to newpath. If newpath already exists and
164
-
// is not a directory, Rename replaces it. OS-specific restrictions may
165
-
// apply when oldpath and newpath are in different directories.
178
+
// Rename renames (moves) oldpath to newpath. If oldpath refers to an
179
+
// in-memory TempFile, its buffer is uploaded to S3 under newpath and the
180
+
// registry entry is dropped — this is how PackWriter's "tmp_pack_… →
181
+
// pack-<sha>.pack" promotion lands the final pack in the bucket. Otherwise
182
+
// Rename uses Tigris's in-place RenameObject extension.
166
183
func (fs3 *S3FS) Rename(oldpath, newpath string) error {
167
184
ctx := context.TODO() // TODO: Get user-supplied context?
168
185
169
186
src := fs3.key(oldpath)
170
187
dst := fs3.key(newpath)
171
188
189
+
if buf, ok := fs3.detachTemp(oldpath); ok {
190
+
data := buf.snapshot()
191
+
_, err := fs3.client.PutObject(ctx, &s3.PutObjectInput{
192
+
Bucket: &fs3.bucket,
193
+
Key: &dst,
194
+
Body: bytes.NewReader(data),
195
+
})
196
+
if err != nil {
197
+
return fmt.Errorf("failed to upload temp %q to %q: %w", oldpath, newpath, err)
198
+
}
199
+
return nil
200
+
}
201
+
172
202
// RenameObject is a Tigris extension that renames in place (no data copy),
173
203
// so we don't need a separate CopyObject + DeleteObject. CopySource is
174
204
// bucket-qualified; Key is the destination key.
···
185
215
return nil
186
216
}
187
217
188
-
// Remove removes the named file or directory.
218
+
// Remove removes the named file or directory. In-memory TempFile entries are
219
+
// dropped from the registry without an S3 call.
189
220
func (fs3 *S3FS) Remove(filename string) error {
190
-
// TODO: Validate the path?
191
-
// ...
221
+
if _, ok := fs3.detachTemp(filename); ok {
222
+
return nil
223
+
}
192
224
193
-
// Create a context
194
225
ctx := context.TODO() // TODO: Get user-supplied context?
195
226
196
227
key := fs3.key(filename)
+1
internal/s3fs/chroot.go
+1
internal/s3fs/chroot.go
+8
internal/s3fs/filesystem.go
+8
internal/s3fs/filesystem.go
···
4
4
"fmt"
5
5
"path"
6
6
"strings"
7
+
"sync"
7
8
8
9
"github.com/go-git/go-billy/v6"
9
10
"github.com/tigrisdata/storage-go"
···
18
19
bucket string
19
20
root string
20
21
separator string
22
+
23
+
// temps holds TempFile-backed buffers keyed by canonical S3 key, so a
24
+
// subsequent Open of the same path returns a reader over the same bytes
25
+
// the writer is still appending to. See tempfs.go.
26
+
tempMu sync.Mutex
27
+
temps map[string]*tempBuffer
21
28
}
22
29
23
30
// NewS3FS creates a new S3FS Filesystem.
···
31
38
bucket: bucket,
32
39
root: "",
33
40
separator: DefaultSeparator,
41
+
temps: make(map[string]*tempBuffer),
34
42
}, nil
35
43
}
36
44
+10
-9
internal/s3fs/tempfile.go
+10
-9
internal/s3fs/tempfile.go
···
10
10
"github.com/go-git/go-billy/v6"
11
11
)
12
12
13
-
// TempFile creates a uniquely named, write-only file under dir whose name
14
-
// begins with prefix. The object is uploaded to S3 when the returned file is
15
-
// closed; until then nothing exists in the bucket. The caller is responsible
16
-
// for renaming or removing it.
17
-
//
18
-
// Note: the returned file is write-only. S3 has no read-while-write temp file,
19
-
// so callers that reopen the temp path for reading before Close (e.g. go-git's
20
-
// streaming PackWriter) are not supported; use the loose-object path instead.
13
+
// TempFile creates a uniquely named file under dir whose name begins with
14
+
// prefix and returns a write handle to it. The bytes live in an in-memory
15
+
// buffer registered against the filesystem; a subsequent Open of the same
16
+
// path returns a reader over that same buffer (needed by go-git's streaming
17
+
// PackWriter, which reads the temp pack back as it is written). The buffer
18
+
// is uploaded to S3 only when the caller renames the path to its final
19
+
// location; Remove discards it.
21
20
func (fs3 *S3FS) TempFile(dir, prefix string) (billy.File, error) {
22
21
var b [16]byte
23
22
if _, err := rand.Read(b[:]); err != nil {
···
25
24
}
26
25
27
26
name := fs3.Join(dir, prefix+hex.EncodeToString(b[:]))
28
-
return newS3WriteFile(fs3.client, fs3.bucket, fs3.key(name), name)
27
+
buf := &tempBuffer{}
28
+
fs3.registerTemp(name, buf)
29
+
return &tempWriteFile{buf: buf, name: name}, nil
29
30
}
+213
internal/s3fs/tempfs.go
+213
internal/s3fs/tempfs.go
···
1
+
// tempfs.go backs billy.TempFile with an in-memory buffer that supports
2
+
// read-while-write on the same path. go-git/v6's streaming PackWriter creates
3
+
// a temp pack file, immediately opens the same path for reading, and reads it
4
+
// back concurrently while writing to build the index. S3 cannot offer that on
5
+
// a single object, so until the final Rename uploads the bytes the buffer is
6
+
// the file.
7
+
8
+
package s3fs
9
+
10
+
import (
11
+
"fmt"
12
+
"io"
13
+
"io/fs"
14
+
"os"
15
+
"sync"
16
+
"time"
17
+
18
+
"github.com/go-git/go-billy/v6"
19
+
)
20
+
21
+
// tempBuffer is a growable byte buffer that one writer and one reader can
22
+
// access concurrently. It is the backing store for a single TempFile entry in
23
+
// the S3FS temp registry.
24
+
type tempBuffer struct {
25
+
mu sync.Mutex
26
+
data []byte
27
+
}
28
+
29
+
func (b *tempBuffer) write(p []byte) (int, error) {
30
+
b.mu.Lock()
31
+
b.data = append(b.data, p...)
32
+
b.mu.Unlock()
33
+
return len(p), nil
34
+
}
35
+
36
+
// readAt copies bytes starting at off. It returns (0, io.EOF) when off is at
37
+
// or past the current end so callers (most notably go-git's syncedReader) can
38
+
// distinguish "no data right now" from a hard error and retry.
39
+
func (b *tempBuffer) readAt(p []byte, off int64) (int, error) {
40
+
if off < 0 {
41
+
return 0, fmt.Errorf("s3fs: negative offset")
42
+
}
43
+
b.mu.Lock()
44
+
defer b.mu.Unlock()
45
+
if off >= int64(len(b.data)) {
46
+
return 0, io.EOF
47
+
}
48
+
return copy(p, b.data[off:]), nil
49
+
}
50
+
51
+
func (b *tempBuffer) size() int64 {
52
+
b.mu.Lock()
53
+
defer b.mu.Unlock()
54
+
return int64(len(b.data))
55
+
}
56
+
57
+
// snapshot returns a copy of the current bytes. Used by Rename to upload the
58
+
// final pack to S3 without holding the mutex during the network call.
59
+
func (b *tempBuffer) snapshot() []byte {
60
+
b.mu.Lock()
61
+
out := make([]byte, len(b.data))
62
+
copy(out, b.data)
63
+
b.mu.Unlock()
64
+
return out
65
+
}
66
+
67
+
// tempWriteFile is the billy.File returned by TempFile. Close marks the handle
68
+
// closed but does not upload; the final Rename uploads to S3 and Remove
69
+
// discards.
70
+
type tempWriteFile struct {
71
+
buf *tempBuffer
72
+
name string
73
+
closed bool
74
+
}
75
+
76
+
func (f *tempWriteFile) Name() string { return f.name }
77
+
78
+
func (f *tempWriteFile) Write(p []byte) (int, error) {
79
+
if f.closed {
80
+
return 0, ErrFileClosed
81
+
}
82
+
return f.buf.write(p)
83
+
}
84
+
85
+
func (f *tempWriteFile) WriteAt(p []byte, off int64) (int, error) {
86
+
return 0, &os.PathError{Op: "write", Path: f.name, Err: ErrNotImplemented}
87
+
}
88
+
89
+
func (f *tempWriteFile) Read(p []byte) (int, error) { return 0, ErrCantReadFromWriteOnly }
90
+
func (f *tempWriteFile) ReadAt(p []byte, off int64) (int, error) { return 0, ErrCantReadFromWriteOnly }
91
+
92
+
func (f *tempWriteFile) Seek(offset int64, whence int) (int64, error) {
93
+
return 0, &os.PathError{Op: "seek", Path: f.name, Err: ErrNotImplemented}
94
+
}
95
+
96
+
func (f *tempWriteFile) Truncate(size int64) error { return ErrTruncateNotSupported }
97
+
func (f *tempWriteFile) Lock() error { return ErrLockNotSupported }
98
+
func (f *tempWriteFile) Unlock() error { return ErrLockNotSupported }
99
+
100
+
func (f *tempWriteFile) Close() error {
101
+
if f.closed {
102
+
return ErrFileClosed
103
+
}
104
+
f.closed = true
105
+
return nil
106
+
}
107
+
108
+
func (f *tempWriteFile) Stat() (fs.FileInfo, error) {
109
+
return newFileInfo(f.name, f.buf.size(), time.Now()), nil
110
+
}
111
+
112
+
// tempReadFile is what Open returns for a path that is still in the temp
113
+
// registry. It carries its own cursor; Read returns (0, io.EOF) at the current
114
+
// end of the buffer so go-git's syncedReader can sleep and retry.
115
+
type tempReadFile struct {
116
+
buf *tempBuffer
117
+
name string
118
+
pos int64
119
+
closed bool
120
+
}
121
+
122
+
func (f *tempReadFile) Name() string { return f.name }
123
+
124
+
func (f *tempReadFile) Read(p []byte) (int, error) {
125
+
if f.closed {
126
+
return 0, ErrFileClosed
127
+
}
128
+
n, err := f.buf.readAt(p, f.pos)
129
+
f.pos += int64(n)
130
+
return n, err
131
+
}
132
+
133
+
func (f *tempReadFile) ReadAt(p []byte, off int64) (int, error) {
134
+
if f.closed {
135
+
return 0, ErrFileClosed
136
+
}
137
+
return f.buf.readAt(p, off)
138
+
}
139
+
140
+
func (f *tempReadFile) Seek(offset int64, whence int) (int64, error) {
141
+
if f.closed {
142
+
return 0, ErrFileClosed
143
+
}
144
+
switch whence {
145
+
case io.SeekStart:
146
+
f.pos = offset
147
+
case io.SeekCurrent:
148
+
f.pos += offset
149
+
case io.SeekEnd:
150
+
f.pos = f.buf.size() + offset
151
+
default:
152
+
return 0, fmt.Errorf("s3fs: invalid whence %d", whence)
153
+
}
154
+
return f.pos, nil
155
+
}
156
+
157
+
func (f *tempReadFile) Write(p []byte) (int, error) { return 0, ErrCantWriteToReadOnly }
158
+
func (f *tempReadFile) WriteAt(p []byte, off int64) (int, error) { return 0, ErrCantWriteToReadOnly }
159
+
func (f *tempReadFile) Truncate(size int64) error { return ErrTruncateNotSupported }
160
+
func (f *tempReadFile) Lock() error { return ErrLockNotSupported }
161
+
func (f *tempReadFile) Unlock() error { return ErrLockNotSupported }
162
+
163
+
func (f *tempReadFile) Close() error {
164
+
if f.closed {
165
+
return ErrFileClosed
166
+
}
167
+
f.closed = true
168
+
return nil
169
+
}
170
+
171
+
func (f *tempReadFile) Stat() (fs.FileInfo, error) {
172
+
return newFileInfo(f.name, f.buf.size(), time.Now()), nil
173
+
}
174
+
175
+
// lookupTemp returns the tempBuffer for a path if it is currently registered,
176
+
// keyed by the canonical S3 key so the lookup matches the key used when
177
+
// inserting.
178
+
func (fs3 *S3FS) lookupTemp(name string) (*tempBuffer, bool) {
179
+
fs3.tempMu.Lock()
180
+
defer fs3.tempMu.Unlock()
181
+
buf, ok := fs3.temps[fs3.key(name)]
182
+
return buf, ok
183
+
}
184
+
185
+
// registerTemp installs buf at the canonical key for name. Used by TempFile.
186
+
func (fs3 *S3FS) registerTemp(name string, buf *tempBuffer) {
187
+
fs3.tempMu.Lock()
188
+
if fs3.temps == nil {
189
+
fs3.temps = make(map[string]*tempBuffer)
190
+
}
191
+
fs3.temps[fs3.key(name)] = buf
192
+
fs3.tempMu.Unlock()
193
+
}
194
+
195
+
// detachTemp removes a path from the registry and returns its buffer, if any.
196
+
// Used by Rename (after which the bytes are uploaded to S3) and Remove (which
197
+
// discards them).
198
+
func (fs3 *S3FS) detachTemp(name string) (*tempBuffer, bool) {
199
+
fs3.tempMu.Lock()
200
+
defer fs3.tempMu.Unlock()
201
+
k := fs3.key(name)
202
+
buf, ok := fs3.temps[k]
203
+
if ok {
204
+
delete(fs3.temps, k)
205
+
}
206
+
return buf, ok
207
+
}
208
+
209
+
// Compile-time assertions: the temp handles satisfy billy.File.
210
+
var (
211
+
_ billy.File = (*tempWriteFile)(nil)
212
+
_ billy.File = (*tempReadFile)(nil)
213
+
)
+223
internal/s3fs/tempfs_test.go
+223
internal/s3fs/tempfs_test.go
···
1
+
package s3fs
2
+
3
+
import (
4
+
"bytes"
5
+
"errors"
6
+
"io"
7
+
"strings"
8
+
"sync"
9
+
"testing"
10
+
)
11
+
12
+
// newTempFS returns an S3FS with only the fields the temp-file code touches.
13
+
// No S3 client is needed because TempFile, Open(temp), Remove(temp), and the
14
+
// read/write handles never reach S3 until Rename uploads.
15
+
func newTempFS() *S3FS {
16
+
return &S3FS{
17
+
bucket: "test",
18
+
separator: DefaultSeparator,
19
+
temps: make(map[string]*tempBuffer),
20
+
}
21
+
}
22
+
23
+
// TestTempFileReadWhileWrite locks in the read-while-write semantics go-git's
24
+
// streaming PackWriter relies on: TempFile + Open of the same path must share
25
+
// a buffer, reads at the current end must return io.EOF (not "not found"),
26
+
// and Seek must let the reader rewind to re-parse from the start.
27
+
func TestTempFileReadWhileWrite(t *testing.T) {
28
+
for _, tt := range []struct {
29
+
name string
30
+
run func(t *testing.T, fs *S3FS, fw, fr io.ReadWriteSeeker)
31
+
}{
32
+
{
33
+
name: "read sees writes",
34
+
run: func(t *testing.T, _ *S3FS, fw, fr io.ReadWriteSeeker) {
35
+
if _, err := fw.Write([]byte("hello world")); err != nil {
36
+
t.Fatalf("Write: %v", err)
37
+
}
38
+
got, err := io.ReadAll(fr)
39
+
if err != nil {
40
+
t.Fatalf("ReadAll: %v", err)
41
+
}
42
+
if string(got) != "hello world" {
43
+
t.Logf("want: %q", "hello world")
44
+
t.Logf("got: %q", string(got))
45
+
t.Error("read did not see written bytes")
46
+
}
47
+
},
48
+
},
49
+
{
50
+
name: "EOF at current end, then resume after more writes",
51
+
run: func(t *testing.T, _ *S3FS, fw, fr io.ReadWriteSeeker) {
52
+
if _, err := fw.Write([]byte("part1")); err != nil {
53
+
t.Fatalf("Write: %v", err)
54
+
}
55
+
buf := make([]byte, 5)
56
+
if n, err := fr.Read(buf); err != nil || n != 5 {
57
+
t.Fatalf("first Read: n=%d err=%v", n, err)
58
+
}
59
+
if n, err := fr.Read(buf); !errors.Is(err, io.EOF) || n != 0 {
60
+
t.Fatalf("Read at end: n=%d err=%v, want (0, io.EOF)", n, err)
61
+
}
62
+
if _, err := fw.Write([]byte("part2")); err != nil {
63
+
t.Fatalf("Write 2: %v", err)
64
+
}
65
+
if n, err := fr.Read(buf); err != nil || n != 5 {
66
+
t.Fatalf("Read after second write: n=%d err=%v", n, err)
67
+
}
68
+
if string(buf) != "part2" {
69
+
t.Errorf("got %q, want %q", string(buf), "part2")
70
+
}
71
+
},
72
+
},
73
+
{
74
+
name: "seek to start re-reads the whole buffer",
75
+
run: func(t *testing.T, _ *S3FS, fw, fr io.ReadWriteSeeker) {
76
+
if _, err := fw.Write([]byte("abcdef")); err != nil {
77
+
t.Fatalf("Write: %v", err)
78
+
}
79
+
if _, err := io.ReadAll(fr); err != nil {
80
+
t.Fatalf("drain: %v", err)
81
+
}
82
+
if pos, err := fr.Seek(0, io.SeekStart); err != nil || pos != 0 {
83
+
t.Fatalf("Seek(0): pos=%d err=%v", pos, err)
84
+
}
85
+
got, err := io.ReadAll(fr)
86
+
if err != nil {
87
+
t.Fatalf("ReadAll after seek: %v", err)
88
+
}
89
+
if string(got) != "abcdef" {
90
+
t.Errorf("got %q, want %q", string(got), "abcdef")
91
+
}
92
+
},
93
+
},
94
+
} {
95
+
t.Run(tt.name, func(t *testing.T) {
96
+
fs := newTempFS()
97
+
fw, err := fs.TempFile("objects/pack", "tmp_pack_")
98
+
if err != nil {
99
+
t.Fatalf("TempFile: %v", err)
100
+
}
101
+
if !strings.HasPrefix(fw.Name(), "objects/pack/tmp_pack_") {
102
+
t.Fatalf("unexpected temp name: %q", fw.Name())
103
+
}
104
+
fr, err := fs.Open(fw.Name())
105
+
if err != nil {
106
+
t.Fatalf("Open(%q): %v", fw.Name(), err)
107
+
}
108
+
tt.run(t, fs, fw, fr)
109
+
})
110
+
}
111
+
}
112
+
113
+
// TestTempFileReadAt covers the io.ReaderAt path that idxfile parsing uses
114
+
// to seek around in the pack while it is being indexed.
115
+
func TestTempFileReadAt(t *testing.T) {
116
+
fs := newTempFS()
117
+
fw, err := fs.TempFile("objects/pack", "tmp_pack_")
118
+
if err != nil {
119
+
t.Fatalf("TempFile: %v", err)
120
+
}
121
+
if _, err := fw.Write([]byte("0123456789")); err != nil {
122
+
t.Fatalf("Write: %v", err)
123
+
}
124
+
fr, err := fs.Open(fw.Name())
125
+
if err != nil {
126
+
t.Fatalf("Open: %v", err)
127
+
}
128
+
buf := make([]byte, 4)
129
+
n, err := fr.ReadAt(buf, 3)
130
+
if err != nil || n != 4 {
131
+
t.Fatalf("ReadAt(_, 3): n=%d err=%v", n, err)
132
+
}
133
+
if string(buf) != "3456" {
134
+
t.Errorf("got %q, want %q", string(buf), "3456")
135
+
}
136
+
// Reading past end returns io.EOF.
137
+
if n, err := fr.ReadAt(make([]byte, 1), 100); !errors.Is(err, io.EOF) || n != 0 {
138
+
t.Errorf("ReadAt past end: n=%d err=%v, want (0, io.EOF)", n, err)
139
+
}
140
+
}
141
+
142
+
// TestTempFileRemove drops the registry entry without hitting S3 — important
143
+
// because a nil S3 client would otherwise crash the test. After Remove, Open
144
+
// of the same path must no longer return the temp buffer.
145
+
func TestTempFileRemove(t *testing.T) {
146
+
fs := newTempFS()
147
+
fw, err := fs.TempFile("objects/pack", "tmp_pack_")
148
+
if err != nil {
149
+
t.Fatalf("TempFile: %v", err)
150
+
}
151
+
if err := fs.Remove(fw.Name()); err != nil {
152
+
t.Fatalf("Remove: %v", err)
153
+
}
154
+
if _, ok := fs.lookupTemp(fw.Name()); ok {
155
+
t.Fatal("temp registry still has entry after Remove")
156
+
}
157
+
if len(fs.temps) != 0 {
158
+
t.Errorf("temps len = %d, want 0", len(fs.temps))
159
+
}
160
+
}
161
+
162
+
// TestTempFileConcurrentWriteRead exercises the actual go-git pattern: one
163
+
// goroutine writes, another reads, and the reader retries on (0, io.EOF) like
164
+
// syncedReader does. The full payload must round-trip.
165
+
func TestTempFileConcurrentWriteRead(t *testing.T) {
166
+
fs := newTempFS()
167
+
fw, err := fs.TempFile("objects/pack", "tmp_pack_")
168
+
if err != nil {
169
+
t.Fatalf("TempFile: %v", err)
170
+
}
171
+
fr, err := fs.Open(fw.Name())
172
+
if err != nil {
173
+
t.Fatalf("Open: %v", err)
174
+
}
175
+
176
+
payload := bytes.Repeat([]byte("xyzpdq"), 4096) // ~24 KiB
177
+
var got bytes.Buffer
178
+
var wg sync.WaitGroup
179
+
180
+
done := make(chan struct{})
181
+
wg.Add(1)
182
+
go func() {
183
+
defer wg.Done()
184
+
buf := make([]byte, 1024)
185
+
for {
186
+
n, err := fr.Read(buf)
187
+
if n > 0 {
188
+
got.Write(buf[:n])
189
+
}
190
+
if errors.Is(err, io.EOF) {
191
+
if got.Len() == len(payload) {
192
+
return
193
+
}
194
+
select {
195
+
case <-done:
196
+
return
197
+
default:
198
+
continue
199
+
}
200
+
}
201
+
if err != nil {
202
+
t.Errorf("Read: %v", err)
203
+
return
204
+
}
205
+
}
206
+
}()
207
+
208
+
for off := 0; off < len(payload); off += 1000 {
209
+
end := off + 1000
210
+
if end > len(payload) {
211
+
end = len(payload)
212
+
}
213
+
if _, err := fw.Write(payload[off:end]); err != nil {
214
+
t.Fatalf("Write: %v", err)
215
+
}
216
+
}
217
+
close(done)
218
+
wg.Wait()
219
+
220
+
if !bytes.Equal(got.Bytes(), payload) {
221
+
t.Errorf("payload mismatch: got %d bytes, want %d", got.Len(), len(payload))
222
+
}
223
+
}
+15
internal/slog.go
+15
internal/slog.go
···
1
+
package internal
2
+
3
+
import (
4
+
"log/slog"
5
+
"os"
6
+
)
7
+
8
+
func InitSlog(level string) (*slog.Logger, error) {
9
+
var lvl slog.Level
10
+
if err := lvl.UnmarshalText([]byte(level)); err != nil {
11
+
return nil, err
12
+
}
13
+
14
+
return slog.New(slog.NewJSONHandler(os.Stderr, &slog.HandlerOptions{Level: lvl})), nil
15
+
}