1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
|
package main
// syncd fronts git-forge: a repository is filled in before it is served and a
// write is durable before its response leaves. A flush walks the whole
// datadir, so a misclassified route costs delay, not durability.
import (
"bytes"
"fmt"
"log"
"net/http"
"net/http/httputil"
"strings"
)
var proxy = &httputil.ReverseProxy{
Rewrite: func(pr *httputil.ProxyRequest) {
pr.Out.URL.Scheme = "http"
pr.Out.URL.Host = forgeAddr
},
// Unbuffered: a clone streams.
FlushInterval: -1,
}
// Asked for by the idle alarm. Counted here: a fetch in the worker resolves at
// the headers, minutes before a clone's last byte.
const inflightPath = "/.syncd/inflight"
// What a held response may weigh. A write's answer is kilobytes; the bound is
// for cgit, which answers a snapshot to POST as well.
const maxHeld = 8 << 20
func (s *Sync) ServeHTTP(w http.ResponseWriter, r *http.Request) {
// RawPath is set only when the path arrived escaped, past the worker's
// check on the prefix: passed on to 404.
if r.URL.Path == inflightPath && r.URL.RawPath == "" {
fmt.Fprintln(w, s.inflight.Load())
return
}
s.inflight.Add(1)
defer s.inflight.Add(-1)
// Reads too. Deleting is the exception: git-forge removes the directories
// without reading them.
if base, ok := s.repoOf(r.URL.Path); ok && !strings.HasSuffix(r.URL.Path, "/settings/delete") {
if err := s.hydrate(base); err != nil {
log.Printf("%s: %v", r.URL.Path, err)
http.Error(w, "this repository could not be fetched from storage\n", http.StatusServiceUnavailable)
return
}
}
if isRead(r) {
proxy.ServeHTTP(w, r)
return
}
// A write: hold the response until it is in the bucket.
held := &buffered{header: http.Header{}}
proxy.ServeHTTP(held, r)
if held.status/100 == 2 || held.status/100 == 3 {
if err := s.flush(); err != nil {
// A 503 for a write already on disk: the disk does not survive. A
// later flush carries it up.
log.Printf("flush for %s: %v", r.URL.Path, err)
http.Error(w, "the change could not be made durable\n", http.StatusServiceUnavailable)
return
}
}
// Flushed first, so the write is durable and only the body is lost; a
// truncated body would be worse than none.
if held.overflow {
log.Printf("%s %s: response past %d bytes, discarded rather than held",
r.Method, r.URL.Path, maxHeld)
http.Error(w, "this response is too large to hold\n", http.StatusServiceUnavailable)
return
}
held.writeTo(w)
}
// isRead is a request that changes nothing that has to survive.
func isRead(r *http.Request) bool {
p := r.URL.Path
if r.Method != http.MethodGet && r.Method != http.MethodHead {
// A clone is a POST, and gigabytes.
return strings.HasSuffix(p, "/git-upload-pack")
}
// The two get routes that write: both edit users.json.
if p == "/confirm-email" || strings.HasPrefix(p, "/oidc/") {
return false
}
return true
}
// repoOf resolves the repository a path names, two components before one as
// git-forge does. Stripping the suffixes is unambiguous because git-forge
// refuses names ending in them.
func (s *Sync) repoOf(p string) (string, bool) {
p = strings.TrimPrefix(p, "/")
parts := strings.SplitN(p, "/", 3)
for n := min(2, len(parts)); n >= 1; n-- {
base := strings.Join(parts[:n], "/")
base = strings.TrimSuffix(base, ".git")
base = strings.TrimSuffix(base, ".issues")
base = strings.TrimSuffix(base, ".wiki")
if s.known(base) {
return base, true
}
}
return "", false
}
// buffered holds a response until it is durable. Past maxHeld it keeps nothing
// and says so.
type buffered struct {
header http.Header
status int
body bytes.Buffer
overflow bool
}
func (b *buffered) Header() http.Header { return b.header }
func (b *buffered) WriteHeader(status int) {
if b.status == 0 {
b.status = status
}
}
// Write reports success even when it keeps nothing: a short write makes the
// proxy abort the connection before ServeHTTP can flush.
func (b *buffered) Write(p []byte) (int, error) {
if b.status == 0 {
b.status = http.StatusOK
}
if b.overflow || b.body.Len()+len(p) > maxHeld {
b.overflow = true
return len(p), nil
}
return b.body.Write(p)
}
func (b *buffered) writeTo(w http.ResponseWriter) {
for k, v := range b.header {
w.Header()[k] = v
}
if b.status == 0 {
b.status = http.StatusOK
}
w.WriteHeader(b.status)
_, _ = w.Write(b.body.Bytes())
}
|