From c59072f57018de5c69b3aaab27061b266377c0d7 Mon Sep 17 00:00:00 2001 From: Xe Iaso Date: Wed, 27 May 2026 22:43:28 -0400 Subject: [PATCH] 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 --- cmd/objgitd/{daemon.go => git_protocol.go} | 18 +- .../{daemon_test.go => git_protocol_test.go} | 0 cmd/objgitd/http.go | 169 +++++++++++++ cmd/objgitd/http_test.go | 124 ++++++++++ cmd/objgitd/main.go | 65 ++++- internal/s3fs/basic.go | 45 +++- internal/s3fs/chroot.go | 1 + internal/s3fs/filesystem.go | 8 + internal/s3fs/tempfile.go | 19 +- internal/s3fs/tempfs.go | 213 +++++++++++++++++ internal/s3fs/tempfs_test.go | 223 ++++++++++++++++++ internal/slog.go | 15 ++ 12 files changed, 870 insertions(+), 30 deletions(-) rename cmd/objgitd/{daemon.go => git_protocol.go} (83%) rename cmd/objgitd/{daemon_test.go => git_protocol_test.go} (100%) create mode 100644 cmd/objgitd/http.go create mode 100644 cmd/objgitd/http_test.go create mode 100644 internal/s3fs/tempfs.go create mode 100644 internal/s3fs/tempfs_test.go create mode 100644 internal/slog.go diff --git a/cmd/objgitd/daemon.go b/cmd/objgitd/git_protocol.go similarity index 83% rename from cmd/objgitd/daemon.go rename to cmd/objgitd/git_protocol.go index 4852240..9c3d9dd 100644 --- a/cmd/objgitd/daemon.go +++ b/cmd/objgitd/git_protocol.go @@ -26,6 +26,22 @@ import ( // It is cleared once the (possibly long) transfer begins. const handshakeTimeout = 30 * time.Second +// streamingStorer wraps a storage.Storer to hide its optional +// storer.PackfileWriter capability. go-git's UpdateObjectStorage drains the +// incoming pack into PackfileWriter via io.CopyBuffer, which only returns on +// io.EOF — fine over HTTP (the request body has a natural EOF) but a deadlock +// over git://, where the client holds the connection open waiting for the +// server's report-status. With PackfileWriter hidden, UpdateObjectStorage +// falls through to Parser.Parse, which knows the end of the pack from the +// pack format itself and never waits for an EOF. +// +// Trade-off: Parser.Parse writes loose objects (one Rename → one S3 PUT each) +// instead of one packfile, so large git:// pushes incur more S3 calls. HTTP +// keeps the fast PackfileWriter path. +type streamingStorer struct { + storage.Storer +} + // daemon serves the git:// (TCP) protocol out of a billy filesystem. type daemon struct { fs billy.Filesystem @@ -119,7 +135,7 @@ func (d *daemon) handle(ctx context.Context, conn net.Conn) error { _, _ = pktline.WriteError(conn, fmt.Errorf("cannot open repository %q", req.Pathname)) return fmt.Errorf("opening %q for push: %w", req.Pathname, err) } - return transport.ReceivePack(ctx, st, r, conn, &transport.ReceivePackRequest{ + return transport.ReceivePack(ctx, streamingStorer{Storer: st}, r, conn, &transport.ReceivePackRequest{ GitProtocol: gitProtocol, }) diff --git a/cmd/objgitd/daemon_test.go b/cmd/objgitd/git_protocol_test.go similarity index 100% rename from cmd/objgitd/daemon_test.go rename to cmd/objgitd/git_protocol_test.go diff --git a/cmd/objgitd/http.go b/cmd/objgitd/http.go new file mode 100644 index 0000000..4e97f3c --- /dev/null +++ b/cmd/objgitd/http.go @@ -0,0 +1,169 @@ +package main + +import ( + "compress/gzip" + "errors" + "fmt" + "io" + "log/slog" + "net/http" + "net/url" + "strings" + + "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" +) + +// ServeHTTP speaks the git smart-HTTP protocol. It dispatches on the URL suffix +// the way git-http-backend does: repository paths are variable-depth (e.g. +// /foo/bar.git) and precede a fixed endpoint suffix, which http.ServeMux's +// wildcards cannot express. +func (d *daemon) ServeHTTP(w http.ResponseWriter, r *http.Request) { + p := r.URL.Path + switch { + case r.Method == http.MethodGet && strings.HasSuffix(p, "/info/refs"): + d.handleInfoRefs(w, r, strings.TrimSuffix(p, "/info/refs")) + case r.Method == http.MethodPost && strings.HasSuffix(p, "/git-upload-pack"): + d.handleRPC(w, r, transport.UploadPackService, strings.TrimSuffix(p, "/git-upload-pack")) + case r.Method == http.MethodPost && strings.HasSuffix(p, "/git-receive-pack"): + d.handleRPC(w, r, transport.ReceivePackService, strings.TrimSuffix(p, "/git-receive-pack")) + default: + http.NotFound(w, r) + } +} + +// handleInfoRefs serves the reference-discovery phase: +// GET /{repo}/info/refs?service=git-(upload|receive)-pack. +func (d *daemon) handleInfoRefs(w http.ResponseWriter, r *http.Request, repoPath string) { + service := r.URL.Query().Get("service") + switch service { + case transport.UploadPackService, transport.ReceivePackService: + default: + http.Error(w, fmt.Sprintf("unsupported service %q", service), http.StatusBadRequest) + return + } + + st, ok := d.resolve(w, service, repoPath) + if !ok { + return + } + + slog.Info("serving smart-http advertisement", + "service", service, + "path", repoPath, + "remote", r.RemoteAddr, + ) + + w.Header().Set("Content-Type", "application/x-"+service+"-advertisement") + w.Header().Set("Cache-Control", "no-cache") + + gitProtocol := r.Header.Get("Git-Protocol") + out := ioutil.WriteNopCloser(w) + + // AdvertiseRefs+StatelessRPC emits the "# service=...\n" smart-reply prefix + // followed by the ref advertisement, then returns without touching a reader. + var err error + switch service { + case transport.UploadPackService: + err = transport.UploadPack(r.Context(), st, nil, out, &transport.UploadPackRequest{ + AdvertiseRefs: true, + StatelessRPC: true, + GitProtocol: gitProtocol, + }) + case transport.ReceivePackService: + err = transport.ReceivePack(r.Context(), st, nil, out, &transport.ReceivePackRequest{ + AdvertiseRefs: true, + StatelessRPC: true, + GitProtocol: gitProtocol, + }) + } + if err != nil { + slog.Error("smart-http advertisement failed", "service", service, "path", repoPath, "err", err) + } +} + +// handleRPC serves a stateless negotiation round: +// POST /{repo}/git-(upload|receive)-pack. +func (d *daemon) handleRPC(w http.ResponseWriter, r *http.Request, service, repoPath string) { + st, ok := d.resolve(w, service, repoPath) + if !ok { + return + } + + body := r.Body + if r.Header.Get("Content-Encoding") == "gzip" { + gz, err := gzip.NewReader(r.Body) + if err != nil { + http.Error(w, "invalid gzip body", http.StatusBadRequest) + return + } + defer gz.Close() + body = gz + } + + slog.Info("serving smart-http rpc", + "service", service, + "path", repoPath, + "remote", r.RemoteAddr, + ) + + w.Header().Set("Content-Type", "application/x-"+service+"-result") + w.Header().Set("Cache-Control", "no-cache") + + // The server commands call Close between negotiation steps; the body and the + // response writer must survive that, so both are wrapped as no-op closers. + in := io.NopCloser(body) + out := ioutil.WriteNopCloser(w) + gitProtocol := r.Header.Get("Git-Protocol") + + var err error + switch service { + case transport.UploadPackService: + err = transport.UploadPack(r.Context(), st, in, out, &transport.UploadPackRequest{ + StatelessRPC: true, + GitProtocol: gitProtocol, + }) + case transport.ReceivePackService: + err = transport.ReceivePack(r.Context(), st, in, out, &transport.ReceivePackRequest{ + StatelessRPC: true, + GitProtocol: gitProtocol, + }) + } + if err != nil { + // The status line is already sent, so this can only be logged. + slog.Error("smart-http rpc failed", "service", service, "path", repoPath, "err", err) + } +} + +// resolve loads the storer for an HTTP request, applying the same rules as the +// git:// handler: anonymous read, push gated by allowPush, and create-on-first- +// push. It writes an HTTP error and returns ok=false when the request cannot +// proceed. +func (d *daemon) resolve(w http.ResponseWriter, service, repoPath string) (storage.Storer, bool) { + if service == transport.ReceivePackService { + if !d.allowPush { + http.Error(w, "push is disabled on this server", http.StatusForbidden) + return nil, false + } + st, err := d.loadOrInit(repoPath) + if err != nil { + slog.Error("opening repository for push", "path", repoPath, "err", err) + http.Error(w, "cannot open repository", http.StatusInternalServerError) + return nil, false + } + return st, true + } + + st, err := d.loader.Load(&url.URL{Path: repoPath}) + if err != nil { + if errors.Is(err, transport.ErrRepositoryNotFound) { + http.Error(w, "repository not found", http.StatusNotFound) + return nil, false + } + slog.Error("loading repository", "path", repoPath, "err", err) + http.Error(w, "cannot open repository", http.StatusInternalServerError) + return nil, false + } + return st, true +} diff --git a/cmd/objgitd/http_test.go b/cmd/objgitd/http_test.go new file mode 100644 index 0000000..5ffcce7 --- /dev/null +++ b/cmd/objgitd/http_test.go @@ -0,0 +1,124 @@ +package main + +import ( + "net/http/httptest" + "os/exec" + "path/filepath" + "strings" + "testing" + + "github.com/go-git/go-billy/v6" + "github.com/go-git/go-billy/v6/memfs" + "github.com/go-git/go-git/v6/plumbing/transport" +) + +// TestSmartHTTP drives a real git client against the smart-HTTP handler over an +// in-memory filesystem, covering push (create-on-demand), the allowPush gate, +// and clone round-trips. +func TestSmartHTTP(t *testing.T) { + if _, err := exec.LookPath("git"); err != nil { + t.Skip("git not installed") + } + + for _, tt := range []struct { + name string + allowPush bool + doPush bool + wantPushErr bool + wantCloneErr bool + }{ + { + name: "push creates repo and clone round-trips", + allowPush: true, + doPush: true, + }, + { + name: "push rejected when disabled", + allowPush: false, + doPush: true, + wantPushErr: true, + wantCloneErr: true, + }, + { + name: "clone of missing repo fails", + allowPush: true, + doPush: false, + wantCloneErr: true, + }, + } { + t.Run(tt.name, func(t *testing.T) { + ts, fs := newHTTPServer(t, tt.allowPush) + remote := ts.URL + "/test.git" + + var srcHead string + if tt.doPush { + work := seedRepo(t) + srcHead = strings.TrimSpace(runGit(t, work, "rev-parse", "HEAD")) + + out, err := tryGit(work, "push", remote, "main") + if tt.wantPushErr { + if err == nil { + t.Fatalf("expected push to be rejected, got success:\n%s", out) + } + } else if err != nil { + t.Fatalf("push failed: %v\n%s", err, out) + } + } + + // The bare repo must exist on disk iff a push was expected to land. + _, statErr := fs.Stat("/test.git/config") + pushLanded := tt.doPush && !tt.wantPushErr + if pushLanded && statErr != nil { + t.Fatalf("expected repo to be created on push, but config missing: %v", statErr) + } + if !pushLanded && statErr == nil { + t.Fatal("repository must not exist when push did not land") + } + + dst := t.TempDir() + out, err := tryGit(dst, "clone", remote, "cloned") + if tt.wantCloneErr { + if err == nil { + t.Fatalf("expected clone to fail, got success:\n%s", out) + } + return + } + if err != nil { + t.Fatalf("clone failed: %v\n%s", err, out) + } + + gotHead := strings.TrimSpace(runGit(t, filepath.Join(dst, "cloned"), "rev-parse", "HEAD")) + if gotHead != srcHead { + t.Logf("want: %s", srcHead) + t.Logf("got: %s", gotHead) + t.Error("cloned HEAD does not match pushed HEAD") + } + }) + } +} + +// newHTTPServer starts an httptest server backed by a fresh in-memory filesystem +// and returns it alongside that filesystem for state assertions. +func newHTTPServer(t *testing.T, allowPush bool) (*httptest.Server, billy.Filesystem) { + t.Helper() + fs := memfs.New() + d := &daemon{ + fs: fs, + loader: transport.NewFilesystemLoader(fs, false), + allowPush: allowPush, + } + ts := httptest.NewServer(d) + t.Cleanup(ts.Close) + return ts, fs +} + +// seedRepo creates a local git repository with one commit and returns its path. +func seedRepo(t *testing.T) string { + t.Helper() + work := t.TempDir() + runGit(t, work, "init", "-b", "main") + runGit(t, work, "config", "user.email", "test@example.com") + runGit(t, work, "config", "user.name", "Test") + runGit(t, work, "commit", "--allow-empty", "-m", "initial") + return work +} diff --git a/cmd/objgitd/main.go b/cmd/objgitd/main.go index 707a62b..c6721b9 100644 --- a/cmd/objgitd/main.go +++ b/cmd/objgitd/main.go @@ -2,23 +2,30 @@ package main import ( "context" + "errors" "flag" + "fmt" "log/slog" "net" + "net/http" "os" "os/signal" "syscall" + "time" "github.com/facebookgo/flagenv" "github.com/go-git/go-git/v6/plumbing/transport" "github.com/tigrisdata/storage-go" + "golang.org/x/sync/errgroup" + "tangled.org/xeiaso.net/objgit/internal" "tangled.org/xeiaso.net/objgit/internal/s3fs" _ "github.com/joho/godotenv/autoload" ) var ( - bind = flag.String("bind", ":9418", "TCP address to listen on for the git:// protocol") + gitBind = flag.String("git-bind", ":9418", "TCP address to listen on for the git:// protocol; empty disables it") + httpBind = flag.String("http-bind", ":8080", "TCP address to listen on for the git smart-HTTP protocol; empty disables it") bucket = flag.String("bucket", "", "Tigris bucket that holds the git repositories") allowPush = flag.Bool("allow-push", false, "allow unauthenticated git-receive-pack (push) requests") slogLevel = flag.String("slog-level", "INFO", "log level (DEBUG, INFO, WARN, ERROR)") @@ -28,18 +35,23 @@ func main() { flagenv.Parse() flag.Parse() - var lvl slog.Level - if err := lvl.UnmarshalText([]byte(*slogLevel)); err != nil { - slog.Error("invalid -slog-level", "value", *slogLevel, "err", err) + logger, err := internal.InitSlog(*slogLevel) + if err != nil { + fmt.Fprintln(os.Stderr, "error initializing logging stack:", err) os.Exit(1) } - slog.SetDefault(slog.New(slog.NewJSONHandler(os.Stderr, &slog.HandlerOptions{Level: lvl}))) + slog.SetDefault(logger) if *bucket == "" { slog.Error("-bucket is required") os.Exit(1) } + if *gitBind == "" && *httpBind == "" { + slog.Error("at least one of -git-bind or -http-bind must be set") + os.Exit(1) + } + ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer cancel() @@ -61,19 +73,46 @@ func main() { allowPush: *allowPush, } - ln, err := net.Listen("tcp", *bind) - if err != nil { - slog.Error("can't listen", "bind", *bind, "err", err) - os.Exit(1) - } - slog.Info("objgitd listening", - "bind", *bind, + "git_bind", *gitBind, + "http_bind", *httpBind, "bucket", *bucket, "allow_push", *allowPush, ) - if err := d.Serve(ctx, ln); err != nil { + g, gCtx := errgroup.WithContext(ctx) + + if *gitBind != "" { + ln, err := net.Listen("tcp", *gitBind) + if err != nil { + slog.Error("can't listen", "git_bind", *gitBind, "err", err) + os.Exit(1) + } + g.Go(func() error { return d.Serve(gCtx, ln) }) + } + + if *httpBind != "" { + ln, err := net.Listen("tcp", *httpBind) + if err != nil { + slog.Error("can't listen", "http_bind", *httpBind, "err", err) + os.Exit(1) + } + srv := &http.Server{Handler: d} + g.Go(func() error { + if err := srv.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) { + return err + } + return nil + }) + g.Go(func() error { + <-gCtx.Done() + shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + return srv.Shutdown(shutdownCtx) + }) + } + + if err := g.Wait(); err != nil { slog.Error("server stopped", "err", err) os.Exit(1) } diff --git a/internal/s3fs/basic.go b/internal/s3fs/basic.go index 3fd9cc1..6f3402c 100644 --- a/internal/s3fs/basic.go +++ b/internal/s3fs/basic.go @@ -3,6 +3,7 @@ package s3fs import ( + "bytes" "context" "errors" "fmt" @@ -10,6 +11,7 @@ import ( "os" "path" "strings" + "time" "github.com/aws/aws-sdk-go-v2/aws" "github.com/aws/aws-sdk-go-v2/service/s3" @@ -66,6 +68,13 @@ func (fs3 *S3FS) OpenFile(filename string, flag int, perm os.FileMode) (billy.Fi return newS3DirFile(key, fs3.bucket, fs3.client), nil } + // A TempFile that has not yet been renamed lives only in memory; serve + // reads from that buffer so go-git's PackWriter can read the pack back + // while it is still being written. + if buf, ok := fs3.lookupTemp(filename); ok { + return &tempReadFile{buf: buf, name: filename}, nil + } + f, err := newS3ReadFile(fs3.client, fs3.bucket, key, filename) if err == nil { return f, nil @@ -118,6 +127,12 @@ func (fs3 *S3FS) Stat(filename string) (os.FileInfo, error) { return newDirInfo("/"), nil } + // A still-open TempFile lives only in memory; report its current size so + // callers that Stat the temp path before Rename see a consistent view. + if buf, ok := fs3.lookupTemp(filename); ok { + return newFileInfo(path.Base(filename), buf.size(), time.Now()), nil + } + ctx := context.TODO() head, err := fs3.client.HeadObject(ctx, &s3.HeadObjectInput{ @@ -160,15 +175,30 @@ func (fs3 *S3FS) Stat(filename string) (os.FileInfo, error) { return nil, &os.PathError{Op: "stat", Path: filename, Err: fs.ErrNotExist} } -// Rename renames (moves) oldpath to newpath. If newpath already exists and -// is not a directory, Rename replaces it. OS-specific restrictions may -// apply when oldpath and newpath are in different directories. +// Rename renames (moves) oldpath to newpath. If oldpath refers to an +// in-memory TempFile, its buffer is uploaded to S3 under newpath and the +// registry entry is dropped — this is how PackWriter's "tmp_pack_… → +// pack-.pack" promotion lands the final pack in the bucket. Otherwise +// Rename uses Tigris's in-place RenameObject extension. func (fs3 *S3FS) Rename(oldpath, newpath string) error { ctx := context.TODO() // TODO: Get user-supplied context? src := fs3.key(oldpath) dst := fs3.key(newpath) + if buf, ok := fs3.detachTemp(oldpath); ok { + data := buf.snapshot() + _, err := fs3.client.PutObject(ctx, &s3.PutObjectInput{ + Bucket: &fs3.bucket, + Key: &dst, + Body: bytes.NewReader(data), + }) + if err != nil { + return fmt.Errorf("failed to upload temp %q to %q: %w", oldpath, newpath, err) + } + return nil + } + // RenameObject is a Tigris extension that renames in place (no data copy), // so we don't need a separate CopyObject + DeleteObject. CopySource is // bucket-qualified; Key is the destination key. @@ -185,12 +215,13 @@ func (fs3 *S3FS) Rename(oldpath, newpath string) error { return nil } -// Remove removes the named file or directory. +// Remove removes the named file or directory. In-memory TempFile entries are +// dropped from the registry without an S3 call. func (fs3 *S3FS) Remove(filename string) error { - // TODO: Validate the path? - // ... + if _, ok := fs3.detachTemp(filename); ok { + return nil + } - // Create a context ctx := context.TODO() // TODO: Get user-supplied context? key := fs3.key(filename) diff --git a/internal/s3fs/chroot.go b/internal/s3fs/chroot.go index f500595..c081c7b 100644 --- a/internal/s3fs/chroot.go +++ b/internal/s3fs/chroot.go @@ -23,6 +23,7 @@ func (fs3 *S3FS) Chroot(path string) (billy.Filesystem, error) { bucket: fs3.bucket, root: p, separator: fs3.separator, + temps: make(map[string]*tempBuffer), } return nfs, nil } diff --git a/internal/s3fs/filesystem.go b/internal/s3fs/filesystem.go index 533e8b9..1db7b48 100644 --- a/internal/s3fs/filesystem.go +++ b/internal/s3fs/filesystem.go @@ -4,6 +4,7 @@ import ( "fmt" "path" "strings" + "sync" "github.com/go-git/go-billy/v6" "github.com/tigrisdata/storage-go" @@ -18,6 +19,12 @@ type S3FS struct { bucket string root string separator string + + // temps holds TempFile-backed buffers keyed by canonical S3 key, so a + // subsequent Open of the same path returns a reader over the same bytes + // the writer is still appending to. See tempfs.go. + tempMu sync.Mutex + temps map[string]*tempBuffer } // NewS3FS creates a new S3FS Filesystem. @@ -31,6 +38,7 @@ func NewS3FS(client *storage.Client, bucket string) (billy.Filesystem, error) { bucket: bucket, root: "", separator: DefaultSeparator, + temps: make(map[string]*tempBuffer), }, nil } diff --git a/internal/s3fs/tempfile.go b/internal/s3fs/tempfile.go index c08ad2e..33ff05a 100644 --- a/internal/s3fs/tempfile.go +++ b/internal/s3fs/tempfile.go @@ -10,14 +10,13 @@ import ( "github.com/go-git/go-billy/v6" ) -// TempFile creates a uniquely named, write-only file under dir whose name -// begins with prefix. The object is uploaded to S3 when the returned file is -// closed; until then nothing exists in the bucket. The caller is responsible -// for renaming or removing it. -// -// Note: the returned file is write-only. S3 has no read-while-write temp file, -// so callers that reopen the temp path for reading before Close (e.g. go-git's -// streaming PackWriter) are not supported; use the loose-object path instead. +// TempFile creates a uniquely named file under dir whose name begins with +// prefix and returns a write handle to it. The bytes live in an in-memory +// buffer registered against the filesystem; a subsequent Open of the same +// path returns a reader over that same buffer (needed by go-git's streaming +// PackWriter, which reads the temp pack back as it is written). The buffer +// is uploaded to S3 only when the caller renames the path to its final +// location; Remove discards it. func (fs3 *S3FS) TempFile(dir, prefix string) (billy.File, error) { var b [16]byte if _, err := rand.Read(b[:]); err != nil { @@ -25,5 +24,7 @@ func (fs3 *S3FS) TempFile(dir, prefix string) (billy.File, error) { } name := fs3.Join(dir, prefix+hex.EncodeToString(b[:])) - return newS3WriteFile(fs3.client, fs3.bucket, fs3.key(name), name) + buf := &tempBuffer{} + fs3.registerTemp(name, buf) + return &tempWriteFile{buf: buf, name: name}, nil } diff --git a/internal/s3fs/tempfs.go b/internal/s3fs/tempfs.go new file mode 100644 index 0000000..6f7c966 --- /dev/null +++ b/internal/s3fs/tempfs.go @@ -0,0 +1,213 @@ +// tempfs.go backs billy.TempFile with an in-memory buffer that supports +// read-while-write on the same path. go-git/v6's streaming PackWriter creates +// a temp pack file, immediately opens the same path for reading, and reads it +// back concurrently while writing to build the index. S3 cannot offer that on +// a single object, so until the final Rename uploads the bytes the buffer is +// the file. + +package s3fs + +import ( + "fmt" + "io" + "io/fs" + "os" + "sync" + "time" + + "github.com/go-git/go-billy/v6" +) + +// tempBuffer is a growable byte buffer that one writer and one reader can +// access concurrently. It is the backing store for a single TempFile entry in +// the S3FS temp registry. +type tempBuffer struct { + mu sync.Mutex + data []byte +} + +func (b *tempBuffer) write(p []byte) (int, error) { + b.mu.Lock() + b.data = append(b.data, p...) + b.mu.Unlock() + return len(p), nil +} + +// readAt copies bytes starting at off. It returns (0, io.EOF) when off is at +// or past the current end so callers (most notably go-git's syncedReader) can +// distinguish "no data right now" from a hard error and retry. +func (b *tempBuffer) readAt(p []byte, off int64) (int, error) { + if off < 0 { + return 0, fmt.Errorf("s3fs: negative offset") + } + b.mu.Lock() + defer b.mu.Unlock() + if off >= int64(len(b.data)) { + return 0, io.EOF + } + return copy(p, b.data[off:]), nil +} + +func (b *tempBuffer) size() int64 { + b.mu.Lock() + defer b.mu.Unlock() + return int64(len(b.data)) +} + +// snapshot returns a copy of the current bytes. Used by Rename to upload the +// final pack to S3 without holding the mutex during the network call. +func (b *tempBuffer) snapshot() []byte { + b.mu.Lock() + out := make([]byte, len(b.data)) + copy(out, b.data) + b.mu.Unlock() + return out +} + +// tempWriteFile is the billy.File returned by TempFile. Close marks the handle +// closed but does not upload; the final Rename uploads to S3 and Remove +// discards. +type tempWriteFile struct { + buf *tempBuffer + name string + closed bool +} + +func (f *tempWriteFile) Name() string { return f.name } + +func (f *tempWriteFile) Write(p []byte) (int, error) { + if f.closed { + return 0, ErrFileClosed + } + return f.buf.write(p) +} + +func (f *tempWriteFile) WriteAt(p []byte, off int64) (int, error) { + return 0, &os.PathError{Op: "write", Path: f.name, Err: ErrNotImplemented} +} + +func (f *tempWriteFile) Read(p []byte) (int, error) { return 0, ErrCantReadFromWriteOnly } +func (f *tempWriteFile) ReadAt(p []byte, off int64) (int, error) { return 0, ErrCantReadFromWriteOnly } + +func (f *tempWriteFile) Seek(offset int64, whence int) (int64, error) { + return 0, &os.PathError{Op: "seek", Path: f.name, Err: ErrNotImplemented} +} + +func (f *tempWriteFile) Truncate(size int64) error { return ErrTruncateNotSupported } +func (f *tempWriteFile) Lock() error { return ErrLockNotSupported } +func (f *tempWriteFile) Unlock() error { return ErrLockNotSupported } + +func (f *tempWriteFile) Close() error { + if f.closed { + return ErrFileClosed + } + f.closed = true + return nil +} + +func (f *tempWriteFile) Stat() (fs.FileInfo, error) { + return newFileInfo(f.name, f.buf.size(), time.Now()), nil +} + +// tempReadFile is what Open returns for a path that is still in the temp +// registry. It carries its own cursor; Read returns (0, io.EOF) at the current +// end of the buffer so go-git's syncedReader can sleep and retry. +type tempReadFile struct { + buf *tempBuffer + name string + pos int64 + closed bool +} + +func (f *tempReadFile) Name() string { return f.name } + +func (f *tempReadFile) Read(p []byte) (int, error) { + if f.closed { + return 0, ErrFileClosed + } + n, err := f.buf.readAt(p, f.pos) + f.pos += int64(n) + return n, err +} + +func (f *tempReadFile) ReadAt(p []byte, off int64) (int, error) { + if f.closed { + return 0, ErrFileClosed + } + return f.buf.readAt(p, off) +} + +func (f *tempReadFile) Seek(offset int64, whence int) (int64, error) { + if f.closed { + return 0, ErrFileClosed + } + switch whence { + case io.SeekStart: + f.pos = offset + case io.SeekCurrent: + f.pos += offset + case io.SeekEnd: + f.pos = f.buf.size() + offset + default: + return 0, fmt.Errorf("s3fs: invalid whence %d", whence) + } + return f.pos, nil +} + +func (f *tempReadFile) Write(p []byte) (int, error) { return 0, ErrCantWriteToReadOnly } +func (f *tempReadFile) WriteAt(p []byte, off int64) (int, error) { return 0, ErrCantWriteToReadOnly } +func (f *tempReadFile) Truncate(size int64) error { return ErrTruncateNotSupported } +func (f *tempReadFile) Lock() error { return ErrLockNotSupported } +func (f *tempReadFile) Unlock() error { return ErrLockNotSupported } + +func (f *tempReadFile) Close() error { + if f.closed { + return ErrFileClosed + } + f.closed = true + return nil +} + +func (f *tempReadFile) Stat() (fs.FileInfo, error) { + return newFileInfo(f.name, f.buf.size(), time.Now()), nil +} + +// lookupTemp returns the tempBuffer for a path if it is currently registered, +// keyed by the canonical S3 key so the lookup matches the key used when +// inserting. +func (fs3 *S3FS) lookupTemp(name string) (*tempBuffer, bool) { + fs3.tempMu.Lock() + defer fs3.tempMu.Unlock() + buf, ok := fs3.temps[fs3.key(name)] + return buf, ok +} + +// registerTemp installs buf at the canonical key for name. Used by TempFile. +func (fs3 *S3FS) registerTemp(name string, buf *tempBuffer) { + fs3.tempMu.Lock() + if fs3.temps == nil { + fs3.temps = make(map[string]*tempBuffer) + } + fs3.temps[fs3.key(name)] = buf + fs3.tempMu.Unlock() +} + +// detachTemp removes a path from the registry and returns its buffer, if any. +// Used by Rename (after which the bytes are uploaded to S3) and Remove (which +// discards them). +func (fs3 *S3FS) detachTemp(name string) (*tempBuffer, bool) { + fs3.tempMu.Lock() + defer fs3.tempMu.Unlock() + k := fs3.key(name) + buf, ok := fs3.temps[k] + if ok { + delete(fs3.temps, k) + } + return buf, ok +} + +// Compile-time assertions: the temp handles satisfy billy.File. +var ( + _ billy.File = (*tempWriteFile)(nil) + _ billy.File = (*tempReadFile)(nil) +) diff --git a/internal/s3fs/tempfs_test.go b/internal/s3fs/tempfs_test.go new file mode 100644 index 0000000..08f7576 --- /dev/null +++ b/internal/s3fs/tempfs_test.go @@ -0,0 +1,223 @@ +package s3fs + +import ( + "bytes" + "errors" + "io" + "strings" + "sync" + "testing" +) + +// newTempFS returns an S3FS with only the fields the temp-file code touches. +// No S3 client is needed because TempFile, Open(temp), Remove(temp), and the +// read/write handles never reach S3 until Rename uploads. +func newTempFS() *S3FS { + return &S3FS{ + bucket: "test", + separator: DefaultSeparator, + temps: make(map[string]*tempBuffer), + } +} + +// TestTempFileReadWhileWrite locks in the read-while-write semantics go-git's +// streaming PackWriter relies on: TempFile + Open of the same path must share +// a buffer, reads at the current end must return io.EOF (not "not found"), +// and Seek must let the reader rewind to re-parse from the start. +func TestTempFileReadWhileWrite(t *testing.T) { + for _, tt := range []struct { + name string + run func(t *testing.T, fs *S3FS, fw, fr io.ReadWriteSeeker) + }{ + { + name: "read sees writes", + run: func(t *testing.T, _ *S3FS, fw, fr io.ReadWriteSeeker) { + if _, err := fw.Write([]byte("hello world")); err != nil { + t.Fatalf("Write: %v", err) + } + got, err := io.ReadAll(fr) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if string(got) != "hello world" { + t.Logf("want: %q", "hello world") + t.Logf("got: %q", string(got)) + t.Error("read did not see written bytes") + } + }, + }, + { + name: "EOF at current end, then resume after more writes", + run: func(t *testing.T, _ *S3FS, fw, fr io.ReadWriteSeeker) { + if _, err := fw.Write([]byte("part1")); err != nil { + t.Fatalf("Write: %v", err) + } + buf := make([]byte, 5) + if n, err := fr.Read(buf); err != nil || n != 5 { + t.Fatalf("first Read: n=%d err=%v", n, err) + } + if n, err := fr.Read(buf); !errors.Is(err, io.EOF) || n != 0 { + t.Fatalf("Read at end: n=%d err=%v, want (0, io.EOF)", n, err) + } + if _, err := fw.Write([]byte("part2")); err != nil { + t.Fatalf("Write 2: %v", err) + } + if n, err := fr.Read(buf); err != nil || n != 5 { + t.Fatalf("Read after second write: n=%d err=%v", n, err) + } + if string(buf) != "part2" { + t.Errorf("got %q, want %q", string(buf), "part2") + } + }, + }, + { + name: "seek to start re-reads the whole buffer", + run: func(t *testing.T, _ *S3FS, fw, fr io.ReadWriteSeeker) { + if _, err := fw.Write([]byte("abcdef")); err != nil { + t.Fatalf("Write: %v", err) + } + if _, err := io.ReadAll(fr); err != nil { + t.Fatalf("drain: %v", err) + } + if pos, err := fr.Seek(0, io.SeekStart); err != nil || pos != 0 { + t.Fatalf("Seek(0): pos=%d err=%v", pos, err) + } + got, err := io.ReadAll(fr) + if err != nil { + t.Fatalf("ReadAll after seek: %v", err) + } + if string(got) != "abcdef" { + t.Errorf("got %q, want %q", string(got), "abcdef") + } + }, + }, + } { + t.Run(tt.name, func(t *testing.T) { + fs := newTempFS() + fw, err := fs.TempFile("objects/pack", "tmp_pack_") + if err != nil { + t.Fatalf("TempFile: %v", err) + } + if !strings.HasPrefix(fw.Name(), "objects/pack/tmp_pack_") { + t.Fatalf("unexpected temp name: %q", fw.Name()) + } + fr, err := fs.Open(fw.Name()) + if err != nil { + t.Fatalf("Open(%q): %v", fw.Name(), err) + } + tt.run(t, fs, fw, fr) + }) + } +} + +// TestTempFileReadAt covers the io.ReaderAt path that idxfile parsing uses +// to seek around in the pack while it is being indexed. +func TestTempFileReadAt(t *testing.T) { + fs := newTempFS() + fw, err := fs.TempFile("objects/pack", "tmp_pack_") + if err != nil { + t.Fatalf("TempFile: %v", err) + } + if _, err := fw.Write([]byte("0123456789")); err != nil { + t.Fatalf("Write: %v", err) + } + fr, err := fs.Open(fw.Name()) + if err != nil { + t.Fatalf("Open: %v", err) + } + buf := make([]byte, 4) + n, err := fr.ReadAt(buf, 3) + if err != nil || n != 4 { + t.Fatalf("ReadAt(_, 3): n=%d err=%v", n, err) + } + if string(buf) != "3456" { + t.Errorf("got %q, want %q", string(buf), "3456") + } + // Reading past end returns io.EOF. + if n, err := fr.ReadAt(make([]byte, 1), 100); !errors.Is(err, io.EOF) || n != 0 { + t.Errorf("ReadAt past end: n=%d err=%v, want (0, io.EOF)", n, err) + } +} + +// TestTempFileRemove drops the registry entry without hitting S3 — important +// because a nil S3 client would otherwise crash the test. After Remove, Open +// of the same path must no longer return the temp buffer. +func TestTempFileRemove(t *testing.T) { + fs := newTempFS() + fw, err := fs.TempFile("objects/pack", "tmp_pack_") + if err != nil { + t.Fatalf("TempFile: %v", err) + } + if err := fs.Remove(fw.Name()); err != nil { + t.Fatalf("Remove: %v", err) + } + if _, ok := fs.lookupTemp(fw.Name()); ok { + t.Fatal("temp registry still has entry after Remove") + } + if len(fs.temps) != 0 { + t.Errorf("temps len = %d, want 0", len(fs.temps)) + } +} + +// TestTempFileConcurrentWriteRead exercises the actual go-git pattern: one +// goroutine writes, another reads, and the reader retries on (0, io.EOF) like +// syncedReader does. The full payload must round-trip. +func TestTempFileConcurrentWriteRead(t *testing.T) { + fs := newTempFS() + fw, err := fs.TempFile("objects/pack", "tmp_pack_") + if err != nil { + t.Fatalf("TempFile: %v", err) + } + fr, err := fs.Open(fw.Name()) + if err != nil { + t.Fatalf("Open: %v", err) + } + + payload := bytes.Repeat([]byte("xyzpdq"), 4096) // ~24 KiB + var got bytes.Buffer + var wg sync.WaitGroup + + done := make(chan struct{}) + wg.Add(1) + go func() { + defer wg.Done() + buf := make([]byte, 1024) + for { + n, err := fr.Read(buf) + if n > 0 { + got.Write(buf[:n]) + } + if errors.Is(err, io.EOF) { + if got.Len() == len(payload) { + return + } + select { + case <-done: + return + default: + continue + } + } + if err != nil { + t.Errorf("Read: %v", err) + return + } + } + }() + + for off := 0; off < len(payload); off += 1000 { + end := off + 1000 + if end > len(payload) { + end = len(payload) + } + if _, err := fw.Write(payload[off:end]); err != nil { + t.Fatalf("Write: %v", err) + } + } + close(done) + wg.Wait() + + if !bytes.Equal(got.Bytes(), payload) { + t.Errorf("payload mismatch: got %d bytes, want %d", got.Len(), len(payload)) + } +} diff --git a/internal/slog.go b/internal/slog.go new file mode 100644 index 0000000..6ba6aa5 --- /dev/null +++ b/internal/slog.go @@ -0,0 +1,15 @@ +package internal + +import ( + "log/slog" + "os" +) + +func InitSlog(level string) (*slog.Logger, error) { + var lvl slog.Level + if err := lvl.UnmarshalText([]byte(level)); err != nil { + return nil, err + } + + return slog.New(slog.NewJSONHandler(os.Stderr, &slog.HandlerOptions{Level: lvl})), nil +} -- 2.51.2