Define independent remote lock objects

2026-09-05 12:27:51 +00:00
parent 14b4fee214
commit be59bff690
4 changed files with 625 additions and 1 deletions
--- a/internal/artifacts/remote_lock.go
+++ b/internal/artifacts/remote_lock.go
@@ -0,0 +1,328 @@
+package artifacts
+
+import (
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"path"
+	"sort"
+	"strings"
+	"time"
+
+	"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
+	"gitea.maximumdirect.net/eric/narratio/internal/artifactpolicy"
+	"gitea.maximumdirect.net/eric/narratio/internal/config"
+)
+
+const (
+	RemoteLockFormatVersion  = 1
+	MaxRemoteLockObjectBytes = 64 << 10
+)
+
+const remoteLocksSegment = "locks"
+
+// RemoteLock records an operator lock on one canonical publish source.
+type RemoteLock struct {
+	FormatVersion int
+	Source        string
+	Reason        string
+	CreatedAt     time.Time
+}
+
+// RemoteLockLookup distinguishes an absent lock from a present validated lock.
+type RemoteLockLookup struct {
+	Lock       RemoteLock
+	Present    bool
+	Generation string
+}
+
+type remoteLockDocument struct {
+	FormatVersion int    `json:"format_version"`
+	Source        string `json:"source"`
+	Reason        string `json:"reason,omitempty"`
+	CreatedAt     string `json:"created_at"`
+}
+
+// NewRemoteLock constructs and validates one lock payload.
+func NewRemoteLock(source, reason string, createdAt time.Time) (RemoteLock, error) {
+	lock := RemoteLock{
+		FormatVersion: RemoteLockFormatVersion,
+		Source:        source,
+		Reason:        reason,
+		CreatedAt:     createdAt,
+	}
+	if err := lock.Validate(); err != nil {
+		return RemoteLock{}, err
+	}
+	return lock, nil
+}
+
+// Validate checks the complete canonical lock payload contract.
+func (l RemoteLock) Validate() error {
+	if l.FormatVersion != RemoteLockFormatVersion {
+		return fmt.Errorf("unsupported remote lock format version %d", l.FormatVersion)
+	}
+	if err := validateRemoteLockSource(l.Source); err != nil {
+		return fmt.Errorf("remote lock source: %w", err)
+	}
+	if strings.TrimSpace(l.Reason) != l.Reason {
+		return fmt.Errorf("remote lock reason must not have surrounding whitespace")
+	}
+	if l.CreatedAt.IsZero() {
+		return fmt.Errorf("remote lock created_at is required")
+	}
+	_, offset := l.CreatedAt.Zone()
+	if offset != 0 {
+		return fmt.Errorf("remote lock created_at must be UTC")
+	}
+	return nil
+}
+
+// EncodeRemoteLock returns the canonical bounded JSON representation of a lock.
+func EncodeRemoteLock(lock RemoteLock) ([]byte, error) {
+	if err := lock.Validate(); err != nil {
+		return nil, err
+	}
+	document := remoteLockDocument{
+		FormatVersion: lock.FormatVersion,
+		Source:        lock.Source,
+		Reason:        lock.Reason,
+		CreatedAt:     lock.CreatedAt.UTC().Format(time.RFC3339Nano),
+	}
+	data, err := json.Marshal(document)
+	if err != nil {
+		return nil, fmt.Errorf("encode remote lock: %w", err)
+	}
+	if len(data) > MaxRemoteLockObjectBytes {
+		return nil, fmt.Errorf("encode remote lock: payload exceeds %d-byte limit", MaxRemoteLockObjectBytes)
+	}
+	return data, nil
+}
+
+// DecodeRemoteLock strictly decodes and validates one bounded lock payload.
+func DecodeRemoteLock(data []byte) (RemoteLock, error) {
+	if len(data) > MaxRemoteLockObjectBytes {
+		return RemoteLock{}, fmt.Errorf("decode remote lock: payload exceeds %d-byte limit", MaxRemoteLockObjectBytes)
+	}
+	var document remoteLockDocument
+	if err := decodeStrictJSON(data, &document); err != nil {
+		return RemoteLock{}, fmt.Errorf("decode remote lock: %w", err)
+	}
+	createdAt, err := time.Parse(time.RFC3339Nano, document.CreatedAt)
+	if err != nil {
+		return RemoteLock{}, fmt.Errorf("decode remote lock created_at: %w", err)
+	}
+	if !strings.HasSuffix(document.CreatedAt, "Z") || createdAt.UTC().Format(time.RFC3339Nano) != document.CreatedAt {
+		return RemoteLock{}, fmt.Errorf("decode remote lock created_at: must be canonical RFC 3339 UTC")
+	}
+	lock := RemoteLock{
+		FormatVersion: document.FormatVersion,
+		Source:        document.Source,
+		Reason:        document.Reason,
+		CreatedAt:     createdAt,
+	}
+	if err := lock.Validate(); err != nil {
+		return RemoteLock{}, fmt.Errorf("decode remote lock: %w", err)
+	}
+	return lock, nil
+}
+
+// RemoteLockKey returns the independent lock key for one canonical source.
+func RemoteLockKey(sessionPrefix, source string) (string, error) {
+	locksPrefix, err := remoteLocksPrefix(sessionPrefix)
+	if err != nil {
+		return "", err
+	}
+	if err := validateRemoteLockSource(source); err != nil {
+		return "", err
+	}
+	return locksPrefix + remoteLockDigest(source) + ".lock", nil
+}
+
+// LookupRemoteLock opens and validates the lock for one source when present.
+func LookupRemoteLock(ctx context.Context, store storage.ObjectStore, sessionPrefix, source string) (RemoteLockLookup, error) {
+	if store == nil {
+		return RemoteLockLookup{}, fmt.Errorf("lookup remote lock: object store is required")
+	}
+	key, err := RemoteLockKey(sessionPrefix, source)
+	if err != nil {
+		return RemoteLockLookup{}, fmt.Errorf("lookup remote lock: %w", err)
+	}
+	return lookupRemoteLockAtKey(ctx, store, key, source)
+}
+
+// ListRemoteLocks validates every object in the reserved lock prefix and sorts
+// the resulting locks by source.
+func ListRemoteLocks(ctx context.Context, store storage.ObjectStore, sessionPrefix string) ([]RemoteLock, error) {
+	if store == nil {
+		return nil, fmt.Errorf("list remote locks: object store is required")
+	}
+	locksPrefix, err := remoteLocksPrefix(sessionPrefix)
+	if err != nil {
+		return nil, fmt.Errorf("list remote locks: %w", err)
+	}
+	objects, err := store.List(ctx, locksPrefix)
+	if err != nil {
+		return nil, fmt.Errorf("list remote locks under %q: %w", locksPrefix, err)
+	}
+	locks := make([]RemoteLock, 0, len(objects))
+	seenSources := make(map[string]struct{}, len(objects))
+	for _, object := range objects {
+		digest, err := remoteLockDigestFromKey(locksPrefix, object.Key)
+		if err != nil {
+			return nil, fmt.Errorf("list remote locks: %w", err)
+		}
+		info, data, err := storage.ReadObjectBounded(ctx, store, object.Key, MaxRemoteLockObjectBytes)
+		if err != nil {
+			return nil, fmt.Errorf("read remote lock %q: %w", object.Key, err)
+		}
+		if info.Key != "" && info.Key != object.Key {
+			return nil, fmt.Errorf("read remote lock %q: store returned key %q", object.Key, info.Key)
+		}
+		lock, err := DecodeRemoteLock(data)
+		if err != nil {
+			return nil, fmt.Errorf("read remote lock %q: %w", object.Key, err)
+		}
+		if actual := remoteLockDigest(lock.Source); actual != digest {
+			return nil, fmt.Errorf("remote lock %q source hash mismatch", object.Key)
+		}
+		if _, duplicate := seenSources[lock.Source]; duplicate {
+			return nil, fmt.Errorf("remote lock source %q is duplicated", lock.Source)
+		}
+		seenSources[lock.Source] = struct{}{}
+		locks = append(locks, lock)
+	}
+	sort.Slice(locks, func(i, j int) bool { return locks[i].Source < locks[j].Source })
+	return locks, nil
+}
+
+// AddRemoteLock conditionally creates one lock. With force, one observed lock
+// generation may be replaced; conflicts are returned for operator retry.
+func AddRemoteLock(ctx context.Context, store storage.ObjectStore, sessionPrefix string, lock RemoteLock, force bool) error {
+	if store == nil {
+		return fmt.Errorf("add remote lock: object store is required")
+	}
+	data, err := EncodeRemoteLock(lock)
+	if err != nil {
+		return fmt.Errorf("add remote lock: %w", err)
+	}
+	key, err := RemoteLockKey(sessionPrefix, lock.Source)
+	if err != nil {
+		return fmt.Errorf("add remote lock: %w", err)
+	}
+	condition := storage.WriteCondition{RequireAbsent: true}
+	if force {
+		current, err := lookupRemoteLockAtKey(ctx, store, key, lock.Source)
+		if err != nil {
+			return fmt.Errorf("add remote lock %q: %w", lock.Source, err)
+		}
+		if current.Present {
+			if strings.TrimSpace(current.Generation) == "" {
+				return fmt.Errorf("add remote lock %q: current generation is required for replacement", lock.Source)
+			}
+			condition = storage.WriteCondition{MatchETag: current.Generation}
+		}
+	}
+	_, err = store.UploadConditional(ctx, bytes.NewReader(data), key, storage.UploadOptions{ContentType: "application/json"}, condition)
+	if err != nil {
+		return fmt.Errorf("add remote lock %q: %w", lock.Source, err)
+	}
+	return nil
+}
+
+// RemoveRemoteLock idempotently removes the lock for one canonical source.
+func RemoveRemoteLock(ctx context.Context, store storage.ObjectStore, sessionPrefix, source string) error {
+	if store == nil {
+		return fmt.Errorf("remove remote lock: object store is required")
+	}
+	key, err := RemoteLockKey(sessionPrefix, source)
+	if err != nil {
+		return fmt.Errorf("remove remote lock: %w", err)
+	}
+	if err := store.Delete(ctx, key); err != nil {
+		return fmt.Errorf("remove remote lock %q: %w", source, err)
+	}
+	return nil
+}
+
+func lookupRemoteLockAtKey(ctx context.Context, store storage.ObjectStore, key, source string) (RemoteLockLookup, error) {
+	info, data, err := storage.ReadObjectBounded(ctx, store, key, MaxRemoteLockObjectBytes)
+	if errors.Is(err, os.ErrNotExist) {
+		return RemoteLockLookup{}, nil
+	}
+	if err != nil {
+		return RemoteLockLookup{}, fmt.Errorf("read object %q: %w", key, err)
+	}
+	lock, err := DecodeRemoteLock(data)
+	if err != nil {
+		return RemoteLockLookup{}, fmt.Errorf("read object %q: %w", key, err)
+	}
+	if lock.Source != source || remoteLockDigest(lock.Source) != strings.TrimSuffix(path.Base(key), ".lock") {
+		return RemoteLockLookup{}, fmt.Errorf("remote lock %q source hash mismatch", key)
+	}
+	return RemoteLockLookup{Lock: lock, Present: true, Generation: info.ETag}, nil
+}
+
+func validateRemoteLockSource(source string) error {
+	classified, err := artifactpolicy.ClassifySource(source)
+	if err != nil || classified.Kind == artifactpolicy.SourceKindPreviousArtifact {
+		return fmt.Errorf("source %q is not a supported publish source", source)
+	}
+	if classified.ID != source {
+		return fmt.Errorf("source %q is not canonical", source)
+	}
+	return nil
+}
+
+func remoteLocksPrefix(sessionPrefix string) (string, error) {
+	if err := validateRemoteLockSessionPrefix(sessionPrefix); err != nil {
+		return "", err
+	}
+	return sessionPrefix + remoteLocksSegment + "/", nil
+}
+
+func validateRemoteLockSessionPrefix(sessionPrefix string) error {
+	if sessionPrefix == "" || strings.TrimSpace(sessionPrefix) != sessionPrefix || strings.Contains(sessionPrefix, `\`) || !strings.HasSuffix(sessionPrefix, "/") {
+		return fmt.Errorf("session prefix %q must be a canonical relative object prefix", sessionPrefix)
+	}
+	withoutSlash := strings.TrimSuffix(sessionPrefix, "/")
+	if err := validateRemoteObjectKey(withoutSlash); err != nil {
+		return fmt.Errorf("session prefix %q: %w", sessionPrefix, err)
+	}
+	parts := strings.Split(withoutSlash, "/")
+	if len(parts) < 4 || parts[len(parts)-4] != config.S3CampaignsSegment || parts[len(parts)-2] != config.S3SessionsSegment {
+		return fmt.Errorf("session prefix %q does not identify a campaign session", sessionPrefix)
+	}
+	if err := ValidateSessionIdentity(parts[len(parts)-3], parts[len(parts)-1]); err != nil {
+		return fmt.Errorf("session prefix %q: %w", sessionPrefix, err)
+	}
+	return nil
+}
+
+func remoteLockDigest(source string) string {
+	sum := sha256.Sum256([]byte(source))
+	return hex.EncodeToString(sum[:])
+}
+
+func remoteLockDigestFromKey(prefix, key string) (string, error) {
+	if err := validateRemoteObjectKey(key); err != nil {
+		return "", fmt.Errorf("invalid remote lock key %q: %w", key, err)
+	}
+	if !strings.HasPrefix(key, prefix) {
+		return "", fmt.Errorf("unexpected object %q outside remote lock prefix %q", key, prefix)
+	}
+	name := strings.TrimPrefix(key, prefix)
+	if strings.Contains(name, "/") || !strings.HasSuffix(name, ".lock") {
+		return "", fmt.Errorf("unexpected object %q in remote lock prefix", key)
+	}
+	digest := strings.TrimSuffix(name, ".lock")
+	if err := validateSHA256(digest); err != nil {
+		return "", fmt.Errorf("unexpected remote lock object %q: %w", key, err)
+	}
+	return digest, nil
+}