Enter git-forge-cloudflareHEADmaster

Runs git-forge on Cloudflare Containers with R2 for durable storage. A
Worker routes every request to one container through a Durable Object.
Inside it, syncd mirrors the data directory to the bucket and restores
it at boot, holding a write until it is durable.

git-forge is neither modified nor vendored: the image builds it from
its own repository at the commit named in forge.rev.

1 files changed, 208 insertions, 0 deletions

diff --git a/r2.go b/r2.go
new file mode 100644
index 0000000..592a49d
--- /dev/null
+++ b/r2.go
@@ -0,0 +1,208 @@
+package main
+
+// The object store, over plain http to a host the worker intercepts on the
+// same machine: no credential, no signing.
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"os"
+	"path/filepath"
+	"time"
+)
+
+type R2 struct {
+	base   string
+	client *http.Client
+}
+
+func newR2(base string) *R2 {
+	return &R2{base: base, client: &http.Client{Timeout: 10 * time.Minute}}
+}
+
+// Small: the peer is on this machine. Retrying a body is sound because putBody
+// sets GetBody.
+const tries = 3
+
+func (r *R2) do(req *http.Request) (*http.Response, error) {
+	var err error
+	for attempt := range tries {
+		if attempt > 0 {
+			time.Sleep(time.Duration(attempt) * 200 * time.Millisecond)
+			if req.GetBody != nil {
+				if req.Body, err = req.GetBody(); err != nil {
+					return nil, err
+				}
+			}
+		}
+		var resp *http.Response
+		if resp, err = r.client.Do(req); err == nil {
+			if resp.StatusCode < 500 {
+				return resp, nil
+			}
+			body, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
+			resp.Body.Close()
+			err = fmt.Errorf("%s %s: %s: %s", req.Method, req.URL.Path, resp.Status, body)
+		}
+	}
+	return nil, err
+}
+
+func (r *R2) url(key string) string {
+	return r.base + "/" + (&url.URL{Path: key}).EscapedPath()
+}
+
+// get reads a whole object into memory; anything large goes through getFile.
+func (r *R2) get(key string) ([]byte, bool, error) {
+	req, err := http.NewRequest(http.MethodGet, r.url(key), nil)
+	if err != nil {
+		return nil, false, err
+	}
+	resp, err := r.do(req)
+	if err != nil {
+		return nil, false, err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode == http.StatusNotFound {
+		return nil, false, nil
+	}
+	if resp.StatusCode != http.StatusOK {
+		return nil, false, fmt.Errorf("get %s: %s", key, resp.Status)
+	}
+	data, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, false, err
+	}
+	return data, true, nil
+}
+
+// getFile streams an object onto disk through a temporary name: a truncated
+// object file would be believed.
+func (r *R2) getFile(key, path string) error {
+	if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil {
+		return err
+	}
+	req, err := http.NewRequest(http.MethodGet, r.url(key), nil)
+	if err != nil {
+		return err
+	}
+	resp, err := r.do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("get %s: %s", key, resp.Status)
+	}
+
+	tmp, err := os.CreateTemp(filepath.Dir(path), ".syncd*")
+	if err != nil {
+		return err
+	}
+	defer os.Remove(tmp.Name())
+	if _, err := io.Copy(tmp, resp.Body); err != nil {
+		tmp.Close()
+		return err
+	}
+	if err := tmp.Close(); err != nil {
+		return err
+	}
+	return os.Rename(tmp.Name(), path)
+}
+
+func (r *R2) put(key string, data []byte) error {
+	return r.putBody(key, int64(len(data)), func() (io.ReadCloser, error) {
+		return io.NopCloser(bytes.NewReader(data)), nil
+	})
+}
+
+// putFile streams a file up: a packfile can be larger than the container's
+// memory.
+func (r *R2) putFile(key, path string) error {
+	info, err := os.Stat(path)
+	if err != nil {
+		return err
+	}
+	return r.putBody(key, info.Size(), func() (io.ReadCloser, error) { return os.Open(path) })
+}
+
+func (r *R2) putBody(key string, length int64, open func() (io.ReadCloser, error)) error {
+	if length == 0 {
+		// http.NoBody, or ContentLength 0 is sent as unknown and the handler
+		// rejects it.
+		open = func() (io.ReadCloser, error) { return http.NoBody, nil }
+	}
+	req, err := http.NewRequest(http.MethodPut, r.url(key), nil)
+	if err != nil {
+		return err
+	}
+	req.ContentLength = length
+	req.GetBody = open
+	if req.Body, err = req.GetBody(); err != nil {
+		return err
+	}
+	resp, err := r.do(req)
+	if err != nil {
+		return err
+	}
+	resp.Body.Close()
+	if resp.StatusCode/100 != 2 {
+		return fmt.Errorf("put %s: %s", key, resp.Status)
+	}
+	return nil
+}
+
+func (r *R2) delete(key string) error {
+	req, err := http.NewRequest(http.MethodDelete, r.url(key), nil)
+	if err != nil {
+		return err
+	}
+	resp, err := r.do(req)
+	if err != nil {
+		return err
+	}
+	resp.Body.Close()
+	if resp.StatusCode/100 != 2 {
+		return fmt.Errorf("delete %s: %s", key, resp.Status)
+	}
+	return nil
+}
+
+// list follows the cursor to the end: every key or an error, never a page.
+func (r *R2) list(prefix string) ([]string, error) {
+	var keys []string
+	cursor := ""
+	for {
+		u := fmt.Sprintf("%s/?prefix=%s&cursor=%s", r.base, url.QueryEscape(prefix), url.QueryEscape(cursor))
+		req, err := http.NewRequest(http.MethodGet, u, nil)
+		if err != nil {
+			return nil, err
+		}
+		resp, err := r.do(req)
+		if err != nil {
+			return nil, err
+		}
+		if resp.StatusCode != http.StatusOK {
+			resp.Body.Close()
+			return nil, fmt.Errorf("list %s: %s", prefix, resp.Status)
+		}
+		var page struct {
+			Keys   []string `json:"keys"`
+			Cursor string   `json:"cursor"`
+		}
+		err = json.NewDecoder(resp.Body).Decode(&page)
+		resp.Body.Close()
+		if err != nil {
+			return nil, fmt.Errorf("list %s: %v", prefix, err)
+		}
+		keys = append(keys, page.Keys...)
+		if page.Cursor == "" {
+			return keys, nil
+		}
+		cursor = page.Cursor
+	}
+}