package api
import (
"context"
"errors"
"fmt"
"net/http"
"regexp"
"strings"
"time"
"craftdesign.group/quarry/internal/ops"
)
// Read your own writes (#228). A client that pushed and then reads, CI
// or an agent or zit after `landing open`, can be answered from the
// index as it was before the push, because ingest folds a moment later.
// It already knows the hash it wrote, so any read takes ?indexed=HASH:
// the read waits, briefly, until the repository's index holds that
// artifact, and answers 425 Too Early with Retry-After when it does not
// in time. No index position is exposed; the hash is the position.
// indexedBound is how long a read waits for its artifact to be folded.
// A fold is normally under a second; past this the client is told to
// come back rather than held.
var indexedBound = 5 * time.Second
var hexPrefix = regexp.MustCompile(`^[0-9a-fA-F]{4,64}$`)
// waitIndexed runs ingest at once, then checks every 100 ms until the
// index holds hash or bound passes.
func (d Deps) waitIndexed(ctx context.Context, repo, hash string, bound time.Duration) (bool, error) {
if d.Repos == nil {
return false, fmt.Errorf("no repository %q: %w", repo, ops.ErrNotFound)
}
r, ok := d.Repos.Get(repo)
if !ok || r.Index == nil {
return false, fmt.Errorf("no repository %q: %w", repo, ops.ErrNotFound)
}
deadline := time.Now().Add(bound)
synced := false
for {
held, err := r.Index.Holds(hash)
if err != nil || held {
return held, err
}
if !synced {
// The push is durable in the store; folding it now beats
// waiting for the next tick.
synced = true
_ = r.synced()
continue
}
if time.Now().After(deadline) {
return false, nil
}
select {
case <-ctx.Done():
return false, ctx.Err()
case <-time.After(100 * time.Millisecond):
}
}
}
// readIndexed handles ?indexed= on a read: it strips the parameter, so
// the op's own decoding never sees it, and waits. It answers false when
// it has already written the response.
func readIndexed(w http.ResponseWriter, req *http.Request, d *ops.Descriptor, canRead CanRead) bool {
q := req.URL.Query()
hash := q.Get("indexed")
if hash == "" || d.Kind != ops.Query {
return true
}
q.Del("indexed")
req.URL.RawQuery = q.Encode()
repo := req.PathValue("repo")
if repo != "" && canRead != nil && !canRead(req.Context(), repo) {
return true // the op answers not found, as for any unseen repository
}
if repo == "" {
writeError(w, http.StatusBadRequest, errors.New("indexed= waits on a repository's index, and this request names none"))
return false
}
if !hexPrefix.MatchString(hash) {
writeError(w, http.StatusBadRequest, errors.New("indexed= takes an artifact hash or a prefix of at least 4 hex digits"))
return false
}
wait := indexedWait.Load()
if wait == nil {
return true
}
held, err := (*wait)(req.Context(), repo, strings.ToLower(hash))
if err != nil {
if errors.Is(err, ops.ErrNotFound) {
writeError(w, http.StatusNotFound, err)
} else {
writeError(w, http.StatusInternalServerError, err)
}
return false
}
if !held {
w.Header().Set("Retry-After", "2")
writeError(w, http.StatusTooEarly, fmt.Errorf("the index of %s has not folded %s yet; retry shortly", repo, hash))
return false
}
return true
}