Blob storage: objects, uploads and durability

// Presign returns a URL path with an expiry and a signature. The signature is an HMAC of the
// method, the path, the content type and the expiry, keyed with a secret only the application
// and the store know. The client can use the URL; it cannot change any signed part.
func Presign(secret []byte, method, path, contentType string, expires time.Time) string {
  exp := strconv.FormatInt(expires.Unix(), 10)
  q := url.Values{"X-Expires": {exp}, "X-Signature": {sign(secret, method, path, contentType, exp)}}
  return path + "?" + q.Encode()
}

func sign(secret []byte, method, path, contentType, exp string) string {
  m := hmac.New(sha256.New, secret)
  fmt.Fprintf(m, "%s\n%s\n%s\n%s", method, path, contentType, exp)
  return hex.EncodeToString(m.Sum(nil))
}

// Verify checks a presigned request at time now: the signature first, then the expiry.
func Verify(secret []byte, method, path, contentType string, q url.Values, now time.Time) error {
  exp := q.Get("X-Expires")
  want := sign(secret, method, path, contentType, exp)
  if !hmac.Equal([]byte(want), []byte(q.Get("X-Signature"))) { // constant time
    return ErrBadSignature
  }
  t, err := strconv.ParseInt(exp, 10, 64)
  if err != nil {
    return fmt.Errorf("expiry %q: %w", exp, ErrBadSignature)
  }
  if now.Unix() > t {
    return ErrExpired
  }
  return nil
}

// CreateUpload starts a multipart upload for key. Nothing is visible under key until Complete.
func (s *ObjectStore) CreateUpload(key string) string {
  s.mu.Lock()
  defer s.mu.Unlock()
  s.nextID++
  id := "u" + strconv.Itoa(s.nextID)
  s.uploads[id] = &upload{key: key, parts: map[int][]byte{}}
  return id
}

// UploadPart stores one part. Parts can arrive in any order and in parallel; sending a part again
// replaces it. The client sends the SHA-256 of the part, and a corrupted part is refused.
func (s *ObjectStore) UploadPart(id string, n int, data []byte, sum string) error {
  if digest(data) != sum {
    return fmt.Errorf("part %d: %w", n, ErrBadChecksum)
  }
  s.mu.Lock()
  defer s.mu.Unlock()
  u, ok := s.uploads[id]
  if !ok {
    return fmt.Errorf("upload %s: %w", id, ErrNoUpload)
  }
  u.parts[n] = slices.Clone(data)
  return nil
}

// ListParts returns the part numbers the store holds, so a client can resume after a crash.
func (s *ObjectStore) ListParts(id string) ([]int, error) {
  s.mu.Lock()
  defer s.mu.Unlock()
  u, ok := s.uploads[id]
  if !ok {
    return nil, fmt.Errorf("upload %s: %w", id, ErrNoUpload)
  }
  nums := make([]int, 0, len(u.parts))
  for n := range u.parts {
    nums = append(nums, n)
  }
  slices.Sort(nums)
  return nums, nil
}

// Complete joins the listed parts in order into one object. The object appears all at once.
func (s *ObjectStore) Complete(id string, parts []int) (Object, error) {
  s.mu.Lock()
  defer s.mu.Unlock()
  u, ok := s.uploads[id]
  if !ok {
    return Object{}, fmt.Errorf("upload %s: %w", id, ErrNoUpload)
  }
  var data []byte
  for i, n := range parts {
    p, ok := u.parts[n]
    if !ok {
      return Object{}, fmt.Errorf("part %d: %w", n, ErrMissingPart)
    }
    if i < len(parts)-1 && len(p) < s.MinPart {
      return Object{}, fmt.Errorf("part %d is %d bytes: %w", n, len(p), ErrSmallPart)
    }
    data = append(data, p...)
  }
  o := Object{Data: data, SHA256: digest(data), ContentType: "application/octet-stream"}
  s.objects[u.key] = o
  delete(s.uploads, id)
  return o, nil
}

// NewCode builds the encoding matrix. A Vandermonde matrix has the property that any K of its
// rows are invertible. Multiplying it by the inverse of its top K rows keeps that property and
// makes the top K rows the identity, so the data shards are stored as they are.
func NewCode(k, m int) (*Code, error) {
  n := k + m
  if k < 1 || m < 0 || n > 256 {
    return nil, fmt.Errorf("code %d+%d: need k >= 1 and k + m <= 256", k, m)
  }
  v := newMatrix(n, k)
  for i := range n {
    for j := range k {
      v[i][j] = gfPow(byte(i), j)
    }
  }
  topInv, err := v[:k].invert()
  if err != nil {
    return nil, fmt.Errorf("code %d+%d: %w", k, m, err)
  }
  return &Code{K: k, M: m, enc: v.mul(topInv)}, nil
}

// Reconstruct fills in the missing (nil) shards. Any K surviving shards are K known rows of the
// encoding matrix times the data, so inverting those K rows recovers the data shards. The lost
// parity shards are then encoded again.
func (c *Code) Reconstruct(shards [][]byte) error {
  var rows matrix
  var have [][]byte
  for i, s := range shards {
    if s != nil && len(rows) < c.K {
      rows = append(rows, c.enc[i])
      have = append(have, s)
    }
  }
  if len(rows) < c.K {
    return fmt.Errorf("%d of %d shards left, need %d: %w", len(rows), c.K+c.M, c.K, ErrTooFewShards)
  }
  dec, err := rows.invert()
  if err != nil {
    return fmt.Errorf("decode: %w", err)
  }
  data := make([][]byte, c.K)
  for j := range c.K {
    data[j] = c.combine(dec[j], have)
  }
  for i := range shards {
    if shards[i] == nil {
      shards[i] = c.combine(c.enc[i], data)
    }
  }
  return nil
}

// Durable works out the yearly chance to lose an object. The model: disks fail independently at
// rate afr a year, and a failed disk is rebuilt within repairHours. An object is lost only if more
// than M of its K+M disks fail inside one repair window.
func Durable(s Scheme, afr, repairHours float64) Durability {
  n := s.K + s.M
  p := afr * repairHours / 8760 // one disk, one window
  var perWindow float64
  for i := s.M + 1; i <= n; i++ {
    perWindow += binom(n, i) * math.Pow(p, float64(i)) * math.Pow(1-p, float64(n-i))
  }
  windows := 8760 / repairHours
  annual := windows * perWindow
  return Durability{
    Overhead:  float64(n) / float64(s.K),
    Tolerates: s.M,
    P:         p,
    Windows:   windows,
    Lead:      binom(n, s.M+1) * math.Pow(p, float64(s.M+1)),
    PerWindow: perWindow,
    Annual:    annual,
    Nines:     -math.Log10(annual),
  }
}