1package authoring23import (4 "bytes"5 "context"6 "crypto/sha256"7 "encoding/hex"8 "encoding/json"9 "errors"10 "fmt"11 "io"12 "os"13 "path/filepath"14 "sort"15)1617const CaptureSetFormat = "atlas-capture-set/v1"1819type CaptureSetOptions struct {20 Offline bool21 MaxRequests int22 MaxBytes int6423}2425// CaptureSet is the atomic evidence unit selected by a build. Its manifest is26// published only after every root and continuation has terminated completely.27type CaptureSet struct {28 Format string `json:"format"`29 ID string `json:"id"`30 Digest string `json:"digest"`31 Adapter string `json:"adapter"`32 Version string `json:"adapterVersion"`33 Kind RequestKind `json:"kind"`34 CapturedAt string `json:"capturedAt"`35 TotalBytes int64 `json:"totalBytes"`36 Roots []CaptureRoot `json:"roots"`37}3839type CaptureRoot struct {40 Acquisition string `json:"acquisition"`41 Pages []Capture `json:"pages"`42 Termination CaptureTermination `json:"termination"`43}4445type CaptureTermination struct {46 Kind string `json:"kind"`47 Returned int64 `json:"returned,omitempty"`48 Matched *int64 `json:"matched,omitempty"`49}5051// AcquireSet returns one complete logical acquisition. Offline selection never52// composes independently-latest pages, so a build cannot observe a torn crawl.53func (cache *Cache) AcquireSet(ctx context.Context, requests []Request, registry AdapterRegistry, options CaptureSetOptions) (CaptureSet, bool, error) {54 roots, err := canonicalSetRequests(requests)55 if err != nil {56 return CaptureSet{}, false, err57 }58 setID := captureSetID(roots)59 if options.Offline {60 set, err := cache.LatestSet(setID)61 if err != nil {62 return CaptureSet{}, false, fmt.Errorf("capture set %s is not cached: %w", roots[0].Source, err)63 }64 if err := validateSetForRequests(set, roots); err != nil {65 return CaptureSet{}, false, err66 }67 return set, true, nil68 }69 if options.MaxRequests <= 0 {70 options.MaxRequests = defaultBuildBudgets.Requests71 }72 if options.MaxBytes <= 0 {73 options.MaxBytes = defaultBuildBudgets.TotalBytes74 }75 adapter, err := registry.Lookup(roots[0].Adapter)76 if err != nil {77 return CaptureSet{}, false, err78 }79 if adapter.Version() != roots[0].AdapterVersion {80 return CaptureSet{}, false, fmt.Errorf("adapter %s version is %s, request requires %s", adapter.Name(), adapter.Version(), roots[0].AdapterVersion)81 }82 set := CaptureSet{83 Format: CaptureSetFormat, ID: setID, Adapter: adapter.Name(),84 Version: adapter.Version(), Kind: roots[0].Kind,85 Roots: make([]CaptureRoot, 0, len(roots)),86 }87 usedRequests := 088 for _, root := range roots {89 captured, count, err := cache.acquireRoot(ctx, root, adapter, options, usedRequests, set.TotalBytes)90 if err != nil {91 return CaptureSet{}, false, fmt.Errorf("capture set %s: %w", root.Source, err)92 }93 usedRequests += count94 for _, page := range captured.Pages {95 set.TotalBytes += page.Body.Length96 if page.CapturedAt > set.CapturedAt {97 set.CapturedAt = page.CapturedAt98 }99 }100 set.Roots = append(set.Roots, captured)101 }102 set.Digest, err = captureSetDigest(set)103 if err != nil {104 return CaptureSet{}, false, err105 }106 data, err := json.Marshal(set)107 if err != nil {108 return CaptureSet{}, false, err109 }110 path, err := cache.captureSetPath(set.ID, set.Digest)111 if err != nil {112 return CaptureSet{}, false, err113 }114 written, err := writeOnce(path, append(data, '\n'))115 if err != nil && !errors.Is(err, errExistingContent) {116 return CaptureSet{}, false, fmt.Errorf("publish capture set: %w", err)117 }118 selected, err := cache.SelectSet(set.ID, set.Digest)119 if err != nil {120 return CaptureSet{}, false, fmt.Errorf("validate capture set: %w", err)121 }122 return selected, !written, nil123}124125func (cache *Cache) acquireRoot(ctx context.Context, root Request, adapter Adapter, options CaptureSetOptions, usedRequests int, usedBytes int64) (CaptureRoot, int, error) {126 captured := CaptureRoot{Acquisition: requestCacheKey(root)}127 current := root128 progress := PageProgress{RootLocator: root.IdentityLocator, SeenIDs: make(map[string]bool)}129 seen := make(map[string]bool)130 for {131 if usedRequests+len(captured.Pages) >= options.MaxRequests {132 return CaptureRoot{}, 0, fmt.Errorf("pagination exceeds request budget %d", options.MaxRequests)133 }134 if seen[current.IdentityLocator] {135 return CaptureRoot{}, 0, fmt.Errorf("pagination continuation cycle at %s", current.IdentityLocator)136 }137 seen[current.IdentityLocator] = true138 capture, _, err := cache.Acquire(ctx, current, false)139 if err != nil {140 return CaptureRoot{}, 0, err141 }142 if capture.Body.Length > options.MaxBytes-usedBytes-rootBytes(captured) {143 return CaptureRoot{}, 0, fmt.Errorf("pagination exceeds capture-set byte budget %d", options.MaxBytes)144 }145 captured.Pages = append(captured.Pages, capture)146 body, err := os.ReadFile(cache.BlobPath(capture.Body))147 if err != nil {148 return CaptureRoot{}, 0, fmt.Errorf("read continuation page: %w", err)149 }150 decision, err := adapter.Next(current, body, &progress)151 if err != nil {152 return CaptureRoot{}, 0, err153 }154 progress.Returned, progress.Matched = decision.Returned, decision.Matched155 if decision.Done {156 captured.Termination = CaptureTermination{Kind: decision.Termination, Returned: decision.Returned, Matched: decision.Matched}157 return captured, len(captured.Pages), nil158 }159 current = decision.Next160 }161}162163func rootBytes(root CaptureRoot) int64 {164 var total int64165 for _, page := range root.Pages {166 total += page.Body.Length167 }168 return total169}170171func canonicalSetRequests(requests []Request) ([]Request, error) {172 if len(requests) == 0 {173 return nil, fmt.Errorf("capture set has no requests")174 }175 roots := append([]Request(nil), requests...)176 sort.Slice(roots, func(i, j int) bool { return requestCacheKey(roots[i]) < requestCacheKey(roots[j]) })177 seen := make(map[string]bool, len(roots))178 for _, request := range roots {179 key := requestCacheKey(request)180 if err := validateSHA256("acquisition", key); err != nil {181 return nil, err182 }183 if request.Adapter == "" || request.AdapterVersion == "" || request.Kind == "" || seen[key] {184 return nil, fmt.Errorf("capture set has an invalid or duplicate request")185 }186 if request.Adapter != roots[0].Adapter || request.AdapterVersion != roots[0].AdapterVersion || request.Kind != roots[0].Kind {187 return nil, fmt.Errorf("capture set mixes adapters, versions, or kinds")188 }189 seen[key] = true190 }191 return roots, nil192}193194func captureSetID(requests []Request) string {195 keys := make([]string, len(requests))196 for index, request := range requests {197 keys[index] = requestCacheKey(request)198 }199 sort.Strings(keys)200 data, _ := json.Marshal(keys)201 digest := sha256.Sum256(data)202 return hex.EncodeToString(digest[:])203}204205func captureSetDigest(set CaptureSet) (string, error) {206 set.Digest = ""207 data, err := json.Marshal(set)208 if err != nil {209 return "", err210 }211 digest := sha256.Sum256(data)212 return hex.EncodeToString(digest[:]), nil213}214215func (cache *Cache) LatestSet(setID string) (CaptureSet, error) {216 dir, err := cache.captureSetDir(setID)217 if err != nil {218 return CaptureSet{}, err219 }220 entries, err := os.ReadDir(dir)221 if err != nil {222 return CaptureSet{}, err223 }224 var sets []CaptureSet225 for _, entry := range entries {226 if entry.IsDir() || filepath.Ext(entry.Name()) != ".json" {227 continue228 }229 set, err := cache.SelectSet(setID, entry.Name()[:len(entry.Name())-len(".json")])230 if err != nil {231 return CaptureSet{}, fmt.Errorf("validate capture set %s: %w", entry.Name(), err)232 }233 sets = append(sets, set)234 }235 if len(sets) == 0 {236 return CaptureSet{}, os.ErrNotExist237 }238 sort.Slice(sets, func(i, j int) bool {239 if sets[i].CapturedAt != sets[j].CapturedAt {240 return sets[i].CapturedAt < sets[j].CapturedAt241 }242 return sets[i].Digest < sets[j].Digest243 })244 return sets[len(sets)-1], nil245}246247func (cache *Cache) SelectSet(setID, digest string) (CaptureSet, error) {248 path, err := cache.captureSetPath(setID, digest)249 if err != nil {250 return CaptureSet{}, err251 }252 data, err := os.ReadFile(path)253 if err != nil {254 return CaptureSet{}, err255 }256 set, err := decodeCaptureSet(data)257 if err != nil {258 return CaptureSet{}, err259 }260 if err := cache.validateCaptureSet(set, setID, digest); err != nil {261 return CaptureSet{}, err262 }263 return set, nil264}265266func decodeCaptureSet(data []byte) (CaptureSet, error) {267 decoder := json.NewDecoder(bytes.NewReader(data))268 decoder.DisallowUnknownFields()269 var set CaptureSet270 if err := decoder.Decode(&set); err != nil {271 return CaptureSet{}, err272 }273 if err := decoder.Decode(&struct{}{}); !errors.Is(err, io.EOF) {274 return CaptureSet{}, fmt.Errorf("capture set has invalid trailing data")275 }276 return set, nil277}278279func (cache *Cache) validateCaptureSet(set CaptureSet, setID, digest string) error {280 if set.Format != CaptureSetFormat || set.ID != setID || set.Digest != digest || set.Adapter == "" || set.Version == "" || len(set.Roots) == 0 {281 return fmt.Errorf("capture set envelope is invalid")282 }283 switch set.Kind {284 case RequestFeatures, RequestRaster, RequestAsset:285 default:286 return fmt.Errorf("capture set kind %q is invalid", set.Kind)287 }288 computed, err := captureSetDigest(set)289 if err != nil || computed != digest {290 return fmt.Errorf("capture set digest does not match its contents")291 }292 var total int64293 previous := ""294 latest := ""295 adapter, err := defaultAdapterRegistry().Lookup(set.Adapter)296 if err != nil || adapter.Version() != set.Version {297 return fmt.Errorf("capture set adapter is unavailable")298 }299 for _, root := range set.Roots {300 if validateSHA256("acquisition", root.Acquisition) != nil || root.Acquisition <= previous || len(root.Pages) == 0 || root.Termination.Kind == "" {301 return fmt.Errorf("capture set root is invalid")302 }303 previous = root.Acquisition304 if root.Pages[0].RequestHash != root.Acquisition {305 return fmt.Errorf("capture set root does not begin with its acquisition")306 }307 rootSource := root.Pages[0].Source308 for _, page := range root.Pages {309 if page.Source != rootSource || page.Adapter != set.Adapter || page.Kind != string(set.Kind) {310 return fmt.Errorf("capture set page envelope differs from its root")311 }312 if err := cache.validateCapture(page, page.RequestHash, page.Body.SHA256); err != nil {313 return err314 }315 if page.Body.Length > int64(^uint64(0)>>1)-total {316 return fmt.Errorf("capture set byte total overflows")317 }318 total += page.Body.Length319 if page.CapturedAt > latest {320 latest = page.CapturedAt321 }322 }323 if err := cache.validateCaptureRoot(set, root, adapter); err != nil {324 return err325 }326 }327 if total != set.TotalBytes {328 return fmt.Errorf("capture set byte total is %d, want %d", total, set.TotalBytes)329 }330 if latest != set.CapturedAt {331 return fmt.Errorf("capture set time does not match its pages")332 }333 return nil334}335336func (cache *Cache) validateCaptureRoot(set CaptureSet, root CaptureRoot, adapter Adapter) error {337 progress := PageProgress{RootLocator: root.Pages[0].Locator, SeenIDs: make(map[string]bool)}338 for index, page := range root.Pages {339 body, err := os.ReadFile(cache.BlobPath(page.Body))340 if err != nil {341 return err342 }343 request := Request{344 Kind: set.Kind, Source: page.Source, Adapter: set.Adapter, AdapterVersion: set.Version,345 Locator: page.Locator, IdentityLocator: page.Locator, MediaType: page.Body.MediaType,346 }347 decision, err := adapter.Next(request, body, &progress)348 if err != nil {349 return fmt.Errorf("validate captured continuation: %w", err)350 }351 progress.Returned, progress.Matched = decision.Returned, decision.Matched352 last := index == len(root.Pages)-1353 if last && (!decision.Done || !terminationMatches(root.Termination, decision)) {354 return fmt.Errorf("capture set termination does not match its final page")355 }356 if !last && (decision.Done || decision.Next.IdentityLocator != root.Pages[index+1].Locator) {357 return fmt.Errorf("capture set page order does not follow recorded continuations")358 }359 }360 return nil361}362363func terminationMatches(termination CaptureTermination, decision PageDecision) bool {364 return termination.Kind == decision.Termination && termination.Returned == decision.Returned && equalOptionalInt64(termination.Matched, decision.Matched)365}366367func validateSetForRequests(set CaptureSet, requests []Request) error {368 if set.ID != captureSetID(requests) || len(set.Roots) != len(requests) {369 return fmt.Errorf("cached capture set does not match the planned requests")370 }371 for index, request := range requests {372 if set.Roots[index].Acquisition != requestCacheKey(request) || set.Adapter != request.Adapter || set.Version != request.AdapterVersion || set.Kind != request.Kind {373 return fmt.Errorf("cached capture set does not match request %s", request.Source)374 }375 }376 return nil377}378379func (cache *Cache) captureSetDir(setID string) (string, error) {380 if err := validateSHA256("capture set", setID); err != nil {381 return "", err382 }383 return filepath.Join(cache.root, "capture-sets", setID[:2], setID), nil384}385386func (cache *Cache) captureSetPath(setID, digest string) (string, error) {387 dir, err := cache.captureSetDir(setID)388 if err != nil {389 return "", err390 }391 if err := validateSHA256("capture set digest", digest); err != nil {392 return "", err393 }394 return filepath.Join(dir, digest+".json"), nil395}