Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 10 additions & 4 deletions .objgit/hooks/receive-pack
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
# -allow-hooks. It does NOT run on a normal `git` server — it only has meaning
# when this repository is served by objgitd.
#
# Execution environment (see CLAUDE.md "Push hooks"):
# Execution environment (see docs/usage/hooks.md):
# * It runs in a kefka virtual shell, NOT a real OS shell. Only kefka's
# built-in commands are available (coreutils: cat, ls, echo, printf, head,
# tail, cut, sort, uniq, wc, sha256sum, grep, ...). There is no PATH to
Expand All @@ -15,9 +15,9 @@
# starts here.
# * /tmp is the only writable location (also $HOME and $TMPDIR). Writing
# anywhere else fails; a redirect into /src aborts the script.
# * The hook runs asynchronously after the push response, so it cannot reject
# a push. Its stdout/stderr and exit status are written to the server log
# only — the person pushing never sees them.
# * The hook runs after refs are updated. It cannot reject a push. Its output
# streams to the pushing client when sideband is available, and its exit
# status is logged. The push waits for the hook to finish.
# * One run happens per created or updated branch. Branch deletions are
# skipped.
#
Expand All @@ -28,9 +28,15 @@
# OBJGIT_BRANCH short branch name, e.g. main
# OBJGIT_OLD_SHA previous tip (all zeros when the branch was created)
# OBJGIT_NEW_SHA new tip
# OBJGIT_ADDED_FILES_JSON, OBJGIT_CHANGED_FILES_JSON,
# OBJGIT_DELETED_FILES_JSON are JSON arrays of net file changes.
# OBJGIT_CHANGES_FILE points to the full JSON object in /tmp.
# git's usual "<old> <new> <ref>" line is also fed on stdin.

echo "push to ${OBJGIT_REPO} ${OBJGIT_REF}: ${OBJGIT_OLD_SHA} -> ${OBJGIT_NEW_SHA}"
echo "file changes:"
cat "${OBJGIT_CHANGES_FILE}"
echo

# /src is the checkout of the new commit. List what landed at the top level.
echo "top-level contents:"
Expand Down
12 changes: 12 additions & 0 deletions buf.gen.yaml
Original file line number Diff line number Diff line change
@@ -0,0 +1,12 @@
version: v2
managed:
enabled: true
override:
- file_option: go_package_prefix
value: github.com/tigrisdata/objgit/gen
plugins:
- remote: buf.build/protocolbuffers/go:v1.36.8
out: gen
opt: paths=source_relative
inputs:
- directory: proto
9 changes: 9 additions & 0 deletions buf.yaml
Original file line number Diff line number Diff line change
@@ -0,0 +1,9 @@
version: v2
modules:
- path: proto
lint:
use:
- STANDARD
breaking:
use:
- WIRE_JSON
2 changes: 1 addition & 1 deletion cmd/objgitd/git_protocol.go
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@ func (d *daemon) authorize(ctx context.Context, req auth.Request) auth.Decision
// repo.
type daemon struct {
// sysFS holds daemon-level state that is not scoped to a repository (the SSH
// host key); repository storage is resolved per request via resolver.
// host key and webhook settings); repository storage is resolved per request.
sysFS billy.Filesystem
resolver repofs.Resolver
authz auth.Authorizer
Expand Down
150 changes: 77 additions & 73 deletions cmd/objgitd/hooks.go
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package main
import (
"bytes"
"context"
"encoding/json"
"errors"
"io"
"log/slog"
Expand All @@ -18,107 +19,60 @@ import (
"github.com/tigrisdata/objgit/internal/kefkash"
"github.com/tigrisdata/objgit/internal/metrics"
"github.com/tigrisdata/objgit/internal/mountfs"
"github.com/tigrisdata/objgit/internal/pushevents"
"github.com/tigrisdata/objgit/internal/treefs"
"github.com/tigrisdata/objgit/internal/webhook"
"mvdan.cc/sh/v3/expand"
"mvdan.cc/sh/v3/interp"
"mvdan.cc/sh/v3/syntax"
"tangled.org/xeiaso.net/kefka/command/registry"
"tangled.org/xeiaso.net/kefka/command/registry/coreutils"
)

// refUpdate records a single branch ref change observed across a receive-pack.
// refUpdate records a single branch ref change applied by a receive-pack.
// A zero Old means the branch was created; a zero New means it was deleted.
type refUpdate struct {
Name plumbing.ReferenceName
Old plumbing.Hash
New plumbing.Hash
}

// snapshotRefs returns the current hash of every branch ref in st. go-git's
// transport.ReceivePack does not report which refs it changed, so we diff a
// snapshot taken before the push against one taken after.
func snapshotRefs(st storage.Storer) (map[plumbing.ReferenceName]plumbing.Hash, error) {
it, err := st.IterReferences()
if err != nil {
return nil, err
}
defer it.Close()

out := map[plumbing.ReferenceName]plumbing.Hash{}
err = it.ForEach(func(r *plumbing.Reference) error {
if r.Type() == plumbing.HashReference && r.Name().IsBranch() {
out[r.Name()] = r.Hash()
}
return nil
})
if err != nil {
return nil, err
}
return out, nil
}

// diffRefs computes the branch ref changes between two snapshots.
func diffRefs(before, after map[plumbing.ReferenceName]plumbing.Hash) []refUpdate {
var updates []refUpdate
for name, newHash := range after {
oldHash, ok := before[name]
switch {
case !ok:
updates = append(updates, refUpdate{Name: name, Old: plumbing.ZeroHash, New: newHash})
case oldHash != newHash:
updates = append(updates, refUpdate{Name: name, Old: oldHash, New: newHash})
}
}
for name, oldHash := range before {
if _, ok := after[name]; !ok {
updates = append(updates, refUpdate{Name: name, Old: oldHash, New: plumbing.ZeroHash})
}
}
return updates
}

// receivePack runs the receive-pack service and, when hooks are enabled, fires
// the repository's receive-pack hook for each updated branch once the push
// succeeds — synchronously, streaming hook output to the client over the
// sideband progress channel (rendered as "remote: " lines) before the response
// stream is closed. st is both what the service writes through and the storer
// used for ref snapshots and hook checkouts — all three transports now share the
// same Scanner-bounded PackfileWriter path (see writePack), so no transport needs
// a capability-hiding wrapper.
// receivePack runs the receive-pack service and dispatches post-receive work
// for the commands whose ref updates succeeded. Hook output streams to the
// client before the response closes. Webhook delivery also finishes before the
// response closes, but cannot reject an already accepted push.
//
// This is also the one place every push funnels through — smart HTTP and SSH
// both land here, and git:// never serves receive-pack at all — so it is where
// the push concurrency cap is applied, via the d.pushes.admit seam.
func (d *daemon) receivePack(ctx context.Context, st storage.Storer, repoPath string, r io.ReadCloser, w io.WriteCloser, req *transport.ReceivePackRequest) error {
if !d.allowHooks {
webhooks, settingsErr := webhook.Load(d.sysFS, repoPath)
if settingsErr != nil {
slog.Error("webhook: load repository settings", "repo", repoPath, "err", settingsErr)
}
if !d.allowHooks && (webhooks == nil || !webhooks.Enabled(repoPath)) {
err := receivePackStreaming(ctx, st, r, w, req, d.pushes.admit, nil)
d.healHEADAfterPush(err, st, repoPath)
return err
}

before, err := snapshotRefs(st)
if err != nil {
slog.Warn("hook: ref snapshot before push failed", "path", repoPath, "err", err)
}

// onUpdated runs after refs are updated and report-status is sent, but
// before the response stream closes, so hook output reaches the client live.
// progress is the sideband band-2 writer, or nil when the client did not
// negotiate sideband (hooks then fall back to logging only).
onUpdated := func(progress io.Writer) {
after, err := snapshotRefs(st)
if err != nil {
slog.Error("hook: ref snapshot after push failed", "path", repoPath, "err", err)
return
}
updates := diffRefs(before, after)
if len(updates) == 0 {
return
// The callback runs after refs change and report-status is attempted. The
// updates are taken from this request's successful commands, so another
// concurrent push cannot be attributed to this one.
onUpdated := func(progress io.Writer, updates []refUpdate, acceptedAt time.Time) {
if d.allowHooks {
var branches []refUpdate
for _, u := range updates {
if u.Name.IsBranch() {
branches = append(branches, u)
}
}
d.runHooks(repoPath, "receive-pack", st, branches, progress)
}
d.runHooks(repoPath, "receive-pack", st, updates, progress)
d.emitPushWebhooks(ctx, st, repoPath, webhooks, updates, acceptedAt)
}

err = receivePackStreaming(ctx, st, r, w, req, d.pushes.admit, onUpdated)
err := receivePackStreaming(ctx, st, r, w, req, d.pushes.admit, onUpdated)
d.healHEADAfterPush(err, st, repoPath)
return err
}
Expand Down Expand Up @@ -185,6 +139,52 @@ func (d *daemon) runHook(repoPath, service string, st storage.Storer, u refUpdat

ctx, cancel := context.WithTimeout(context.Background(), d.hookTimeout)
defer cancel()
changes, err := pushevents.Diff(ctx, st, u.Old, u.New)
if err != nil {
log.Error("hook: diff changed files", "err", err)
return
}
// Keep empty lists as JSON arrays. Paths can contain whitespace or newlines,
// so JSON is the only unambiguous representation in environment variables.
added := append([]string{}, changes.GetAdded()...)
changed := append([]string{}, changes.GetChanged()...)
deleted := append([]string{}, changes.GetDeleted()...)
addedJSON, err := json.Marshal(added)
if err != nil {
log.Error("hook: encode added files", "err", err)
return
}
changedJSON, err := json.Marshal(changed)
if err != nil {
log.Error("hook: encode changed files", "err", err)
return
}
deletedJSON, err := json.Marshal(deleted)
if err != nil {
log.Error("hook: encode deleted files", "err", err)
return
}
metadata, err := json.Marshal(struct {
Added []string `json:"added"`
Changed []string `json:"changed"`
Deleted []string `json:"deleted"`
}{added, changed, deleted})
if err != nil {
log.Error("hook: encode changed files", "err", err)
return
}
const changesFile = "/tmp/objgit-changes.json"
f, err := fsys.Create(changesFile)
if err != nil {
log.Error("hook: create changes file", "err", err)
return
}
_, writeErr := io.Copy(f, bytes.NewReader(metadata))
closeErr := f.Close()
if err := errors.Join(writeErr, closeErr); err != nil {
log.Error("hook: write changes file", "err", err)
return
}

// When the client negotiated sideband, stream stdout+stderr straight to it
// ("remote: " lines); otherwise buffer for the log. git does not distinguish
Expand Down Expand Up @@ -216,6 +216,10 @@ func (d *daemon) runHook(repoPath, service string, st storage.Storer, u refUpdat
"OBJGIT_BRANCH="+u.Name.Short(),
"OBJGIT_OLD_SHA="+u.Old.String(),
"OBJGIT_NEW_SHA="+u.New.String(),
"OBJGIT_ADDED_FILES_JSON="+string(addedJSON),
"OBJGIT_CHANGED_FILES_JSON="+string(changedJSON),
"OBJGIT_DELETED_FILES_JSON="+string(deletedJSON),
"OBJGIT_CHANGES_FILE="+changesFile,
)
// Mirror git's post-receive stdin: "<old> <new> <ref>\n".
stdin := strings.NewReader(u.Old.String() + " " + u.New.String() + " " + u.Name.String() + "\n")
Expand Down
Loading
Loading