// mirrorattach.go — a Fossil mirror's attachments, fetched into its
// storage (#273).
//
// zit carries no attachment content: a mirror holds an attachment's
// bytes only when they arrived through the sync as an ordinary
// artifact, and on tcl 248 of 3,297 never did. When the repository has
// storage (its own bucket, or the forge's), quarry fetches each such
// file from the upstream's /raw/HASH, keeps it only when it hashes to
// the A card's src, and stores it in the bucket by SHA-256 like any
// upload. The record (internal/oltp/mirrorattach.go) is what serves it
// afterwards: attachment.head and attachment.link answer a fetched file
// as an external one.
//
// Two things ask for a fetch. A page that shows a file the store lacks
// queues that one file, so what readers look at arrives first. After
// each pull that succeeds, a backfill checks a bounded number of the
// mirror's attachments it has not settled yet, so the rest arrive over
// a few pulls without a burst against the upstream. One fetch runs at a
// time across the forge.
package api
import (
"context"
"crypto/sha256"
"encoding/hex"
"errors"
"fmt"
"io"
"log"
"net/http"
"sync"
"time"
"craftdesign.group/quarry/internal/attach"
"craftdesign.group/quarry/internal/index"
"craftdesign.group/quarry/internal/oltp"
"craftdesign.group/quarry/internal/s3"
"craftdesign.group/quarry/internal/uv"
"craftdesign.group/quarry/internal/zitcli"
)
// MirrorAttachStore records what became of each mirrored attachment.
type MirrorAttachStore interface {
MirrorAttachment(repo, src string) (oltp.MirrorAttach, bool, error)
RecordMirrorAttachment(m oltp.MirrorAttach) error
MirrorAttachSettled(repo string, cutoff time.Time) (map[string]bool, error)
}
const (
// mirrorAttachBatch is how many unsettled attachments one backfill
// pass examines: tcl's 3,297 take a few days of hourly pulls.
mirrorAttachBatch = 50
// mirrorAttachRetry is how long a failed fetch rests before a
// backfill tries it again.
mirrorAttachRetry = 24 * time.Hour
// mirrorAttachPace spaces fetches in a backfill, for the upstream.
mirrorAttachPace = 500 * time.Millisecond
// mirrorAttachTimeout bounds one fetch and its upload.
mirrorAttachTimeout = 2 * time.Minute
)
var (
mirrorAttachSlot = make(chan struct{}, 1)
mirrorAttachPending sync.Map // repo + "\x00" + src, or repo for a backfill
mirrorAttachRuns sync.WaitGroup
mirrorAttachMu sync.Mutex
mirrorAttachClosed bool
)
// CloseMirrorAttach refuses new fetches and waits for those running,
// before the forge closes its database.
func CloseMirrorAttach() {
mirrorAttachMu.Lock()
mirrorAttachClosed = true
mirrorAttachMu.Unlock()
mirrorAttachRuns.Wait()
}
// mirroredRow is src as an external attachment when a fetch stored it,
// so the head and the link serve it from the bucket.
func (d Deps) mirroredRow(repo, src, name string) (index.Attachment, bool) {
if d.MirrorAttach == nil {
return index.Attachment{}, false
}
m, ok, err := d.MirrorAttach.MirrorAttachment(repo, src)
if err != nil || !ok || m.State != oltp.MirrorAttachFetched {
return index.Attachment{}, false
}
return index.Attachment{Src: src, Name: name, External: true, SHA256: m.SHA256, Size: m.Size, Mimetype: m.Mimetype}, true
}
// mirrorAttachReady says whether repo can fetch its attachments: a
// Fossil upstream to fetch from and storage to put them in.
func (d Deps) mirrorAttachReady(repo string) bool {
if d.MirrorAttach == nil || d.upstreamRaw(repo, "x") == "" {
return false
}
_, _, _, err := d.boundBucket(repo)
return err == nil
}
// queueMirrorAttach fetches src for repo off the caller's goroutine,
// once however many pages ask.
func (d Deps) queueMirrorAttach(repo, src string) {
if !d.mirrorAttachReady(repo) {
return
}
d.mirrorAttachGo(repo+"\x00"+src, func(ctx context.Context) {
if err := d.fetchMirrorAttach(ctx, repo, src); err != nil {
log.Printf("mirror: %s: attachment %.12s: %v", repo, src, err)
}
})
}
// startMirrorAttachBackfill examines up to mirrorAttachBatch of repo's
// unsettled attachments, after a pull that succeeded.
func (d Deps) startMirrorAttachBackfill(repo string) {
if !d.mirrorAttachReady(repo) {
return
}
d.mirrorAttachGo(repo, func(ctx context.Context) {
if n, err := d.mirrorAttachBackfill(ctx, repo); err != nil {
log.Printf("mirror: %s: attachment backfill after %d: %v", repo, n, err)
}
})
}
// mirrorAttachGo runs fn in the forge's one fetch slot, unless work
// under the same key is already queued.
func (d Deps) mirrorAttachGo(key string, fn func(context.Context)) {
stop := d.Shutdown
if stop == nil {
stop = context.Background()
}
mirrorAttachMu.Lock()
defer mirrorAttachMu.Unlock()
if mirrorAttachClosed || stop.Err() != nil {
return
}
if _, queued := mirrorAttachPending.LoadOrStore(key, true); queued {
return
}
mirrorAttachRuns.Add(1)
go func() {
defer mirrorAttachRuns.Done()
defer mirrorAttachPending.Delete(key)
select {
case mirrorAttachSlot <- struct{}{}:
case <-stop.Done():
return
}
defer func() { <-mirrorAttachSlot }()
if stop.Err() == nil {
fn(stop)
}
}()
}
// mirrorAttachBackfill is one pass: each unsettled src the store holds
// is recorded as held, and each it lacks is fetched.
func (d Deps) mirrorAttachBackfill(ctx context.Context, name string) (int, error) {
repo, err := d.repo(name)
if err != nil {
return 0, err
}
srcs, err := repo.Index.LiveAttachmentSrcs()
if err != nil {
return 0, err
}
settled, err := d.MirrorAttach.MirrorAttachSettled(name, time.Now().Add(-mirrorAttachRetry))
if err != nil {
return 0, err
}
n := 0
for _, src := range srcs {
if n >= mirrorAttachBatch || ctx.Err() != nil {
break
}
if settled[src] {
continue
}
n++
_, serr := repo.Zit.ArtifactSize(ctx, src)
switch {
case serr == nil:
if err := d.MirrorAttach.RecordMirrorAttachment(oltp.MirrorAttach{Repo: name, Src: src, State: oltp.MirrorAttachStore}); err != nil {
return n, err
}
continue
case !errors.Is(serr, zitcli.ErrNoArtifact):
continue // the store could not say; the next pass asks again
}
if err := d.fetchMirrorAttach(ctx, name, src); err != nil {
log.Printf("mirror: %s: attachment %.12s: %v", name, src, err)
}
select {
case <-ctx.Done():
case <-time.After(mirrorAttachPace):
}
}
return n, nil
}
// fetchMirrorAttach fetches src from repo's upstream into its bucket and
// records the outcome. It returns an error only when nothing could be
// recorded; a refusal from the upstream or the bucket is the row's.
func (d Deps) fetchMirrorAttach(ctx context.Context, repo, src string) error {
if m, ok, err := d.MirrorAttach.MirrorAttachment(repo, src); err == nil && ok && m.State == oltp.MirrorAttachFetched {
return nil
}
raw := d.upstreamRaw(repo, src)
if raw == "" {
return nil
}
_, b, bucket, err := d.boundBucket(repo)
if err != nil {
return nil
}
ctx, cancel := context.WithTimeout(ctx, mirrorAttachTimeout)
defer cancel()
fail := func(why string) error {
return d.MirrorAttach.RecordMirrorAttachment(oltp.MirrorAttach{Repo: repo, Src: src, State: oltp.MirrorAttachFailed, Err: why})
}
body, why := fetchUpstreamRaw(ctx, raw, b.LimitBytes)
if why != "" {
return fail(why)
}
if !uv.Verify(src, body) {
return fail("the upstream's bytes do not hash to " + src)
}
sum := sha256.Sum256(body)
sha := hex.EncodeToString(sum[:])
mt := http.DetectContentType(body)
if err := bucket.PutWith(ctx, &http.Client{Timeout: mirrorAttachTimeout}, attach.Key(sha), body,
s3.PutOptions{ContentType: mt}, time.Now()); err != nil {
return fail("writing to the bucket: " + err.Error())
}
return d.MirrorAttach.RecordMirrorAttachment(oltp.MirrorAttach{Repo: repo, Src: src, State: oltp.MirrorAttachFetched,
SHA256: sha, Size: int64(len(body)), Mimetype: mt})
}
// fetchUpstreamRaw reads u whole, refusing more than limit bytes; why
// says in words what went wrong.
func fetchUpstreamRaw(ctx context.Context, u string, limit int64) ([]byte, string) {
req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
if err != nil {
return nil, err.Error()
}
resp, err := http.DefaultClient.Do(req)
if err != nil {
return nil, "the upstream could not be reached: " + err.Error()
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return nil, fmt.Sprintf("the upstream answered %d", resp.StatusCode)
}
if limit <= 0 {
limit = maxAttachmentBytes
}
body, err := io.ReadAll(io.LimitReader(resp.Body, limit+1))
if err != nil {
return nil, "reading the upstream's answer: " + err.Error()
}
if int64(len(body)) > limit {
return nil, fmt.Sprintf("larger than this bucket's limit of %d bytes", limit)
}
return body, ""
}