amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package park
2
3import (
4 "bytes"
5 "context"
6 "encoding/hex"
7 "encoding/json"
8 "errors"
9 "io"
10 "net/url"
11 "reflect"
12 "sort"
13 "strings"
14 "time"
15
16 "github.com/carlosframework/platform/internal/hostclaim"
17 "github.com/carlosframework/platform/internal/lease"
18 "github.com/carlosframework/platform/internal/release"
19)
20
21const SignatureAge = 15 * time.Minute
22
23var (
24 ErrPending = errors.New("park: waiting for verification after horizon")
25 ErrUnsafe = errors.New("park: unsafe or incomplete input")
26 ErrChanged = errors.New("park: authority changed")
27)
28
29type Ref struct{ Key, VersionID string }
30
31func ref(v lease.ObjectVersion) Ref { return Ref{v.Key, v.VersionID} }
32
33type Copy struct {
34 Source lease.ObjectVersion
35 Destination Ref
36}
37type Manifest struct {
38 V int
39 Host, Obligation, Op, Retirement, Region, Location, Incarnation, Generation string
40 Archive []Copy
41 Delete []lease.ObjectVersion
42}
43type Work struct{ Host, ID, Op string }
44
45// Close stops renewal; Release conditionally removes only the held lock version.
46// Acquire must renew until Close, and Check must refuse expired/lost leases.
47type HeldLease interface {
48 Check(context.Context) error
49 Release(context.Context) error
50 Close()
51}
52type Leases interface {
53 Acquire(context.Context, string, string) (HeldLease, error)
54}
55
56type Engine struct {
57 HomeRegion string
58 Store lease.VersionedStore
59 Claims release.Store
60 Bucket, Region string
61 Leases Leases
62 Now func() time.Time
63 NewID func() string
64 // View is supplied by moves/Plan 7b, using precisely this fresh observation.
65 View func(context.Context, hostclaim.Observed) (View, error)
66 // Authorize checks phase-specific stop evidence/fences and chore ownership.
67 // It must not manufacture stop evidence from a stopped parent alone.
68 Authorize func(context.Context, hostclaim.Observed, hostclaim.Obligation) error
69 // VerifyMovesSweep supplies the precondition for listing-time moves evidence.
70 VerifyMovesSweep func(context.Context, hostclaim.Observed, []lease.ObjectVersion) (MoveSweepProof, error)
71 // After is a test crash point, called after durable effects. Nil in callers.
72 After func(string) error
73}
74
75func (e Engine) after(step string) error {
76 if e.After != nil {
77 return e.After(step)
78 }
79 return nil
80}
81func token(s string) bool {
82 return s != "" && s != "." && s != ".." && !strings.ContainsAny(s, "/\\\x00")
83}
84func hex64(s string) bool { _, err := hex.DecodeString(s); return len(s) == 64 && err == nil }
85func find(c *hostclaim.Claim, w Work) (hostclaim.Obligation, bool) {
86 if c == nil {
87 return hostclaim.Obligation{}, false
88 }
89 for _, o := range c.Obligations {
90 if o.ID == w.ID && o.Op == w.Op {
91 return o, true
92 }
93 }
94 return hostclaim.Obligation{}, false
95}
96func validateScope(host string, o hostclaim.Obligation) error {
97 p := o.Park
98 if p.Validate(host) != nil || !token(o.Retirement) || o.Region == "" || o.Owner != "region:"+o.Region {
99 return ErrUnsafe
100 }
101 if o.Kind != "park-replica" && o.Kind != "park-generation" {
102 return ErrUnsafe
103 }
104 if o.Kind == "park-generation" && o.Generation == "" {
105 return ErrUnsafe
106 }
107 return nil
108}
109
110// Bind persists recovery inputs before the first data-bucket effect. Existing
111// Plan 1 chores have no Park field; callers supply real stop-evidence time and
112// a deployment-wide upper bound enforced by every relevant request sender.
113func (e Engine) Bind(ctx context.Context, w Work, p hostclaim.ParkScope) error {
114 obs, err := hostclaim.Read(ctx, e.Claims, w.Host)
115 if err != nil {
116 return err
117 }
118 o, ok := find(obs.Claim, w)
119 if !ok {
120 return ErrChanged
121 }
122 if o.ParkTarget != "" && o.ParkTarget != p.Location {
123 return ErrUnsafe
124 }
125 if o.Park != nil {
126 if o.Park.Location == p.Location && o.Park.Incarnation == p.Incarnation && o.Park.RequestLimit == p.RequestLimit && o.Park.EvidenceAt.Equal(p.EvidenceAt) {
127 return nil
128 }
129 return ErrChanged
130 }
131 if obs.Claim.Holder == nil || (p.Location != obs.Claim.Holder.Location && !isMovesScope(w.Host, p.Location)) || p.Incarnation != obs.Claim.Holder.Incarnation {
132 return ErrUnsafe
133 }
134 o.Park = &p
135 if err := validateScope(w.Host, o); err != nil {
136 return err
137 }
138 if e.Authorize == nil {
139 return ErrUnsafe
140 }
141 if err := e.Authorize(ctx, obs, o); err != nil {
142 return err
143 }
144 _, err = hostclaim.Update(ctx, e.Claims, obs, func(c *hostclaim.Claim) error {
145 for i := range c.Obligations {
146 if c.Obligations[i].ID == w.ID && c.Obligations[i].Op == w.Op {
147 c.Obligations[i].Park = &p
148 return nil
149 }
150 }
151 return ErrChanged
152 })
153 return err
154}
155func isMovesScope(host, location string) bool {
156 base := "moves/" + host
157 return location == base || strings.HasPrefix(location, base+"/")
158}
159func prefixes(location string) []string {
160 return []string{location + "/"}
161}
162
163// locationOf partitions legacy level/generation children from hex incarnation
164// children. No retired-incarnations lookup.
165func locationOf(host, key string) (string, error) {
166 for _, root := range []string{"instances/", "moves/"} {
167 base := root + host + "/"
168 if !strings.HasPrefix(key, base) {
169 continue
170 }
171 tail := strings.TrimPrefix(key, base)
172 if tail == "" {
173 return "", ErrUnsafe
174 }
175 if root == "moves/" {
176 return root + host, nil
177 }
178 first, _, child := strings.Cut(tail, "/")
179 if child && hex64(first) {
180 return "instances/" + host + "/" + first, nil
181 }
182 return "instances/" + host, nil
183 }
184 return "", ErrUnsafe
185}
186func scopeContains(host, location, key string) (bool, error) {
187 if isMovesScope(host, location) {
188 return strings.HasPrefix(key, location+"/"), nil
189 }
190 loc, err := locationOf(host, key)
191 return loc == location, err
192}
193
194// isBuildState reports whether key is the build state beside the lease at
195// location (release.BuildStateKey; ruling R-PARK). At the live holder
196// location it is the marker's twin: live state that belongs to no
197// generation, passed over wherever the marker key is, and a whole-location
198// park there refuses because of it as it refuses because of the marker.
199// Callers ask it only of the holder's location; at a retired location, or
200// in a moves lane, the same file name is residue like any other object.
201func isBuildState(location, key string) bool {
202 return key == release.BuildStateKey(location)
203}
204func isLock(host, location, key string) bool {
205 for _, p := range prefixes(location) {
206 if key == p+"lock.json" {
207 return true
208 }
209 }
210 return false
211}
212func (e Engine) list(ctx context.Context, host, location string) ([]lease.ObjectVersion, error) {
213 var out []lease.ObjectVersion
214 for _, p := range prefixes(location) {
215 vs, err := e.Store.ListVersions(ctx, e.Bucket, p)
216 if err != nil {
217 return nil, err
218 }
219 out = append(out, vs...)
220 }
221 seen := map[Ref]bool{}
222 filtered := out[:0]
223 for _, v := range out {
224 inScope, err := scopeContains(host, location, v.Key)
225 if err != nil || v.VersionID == "" || seen[ref(v)] {
226 return nil, ErrUnsafe
227 }
228 seen[ref(v)] = true
229 if inScope && !isLock(host, location, v.Key) {
230 filtered = append(filtered, v)
231 }
232 }
233 sort.Slice(filtered, func(i, j int) bool {
234 a, b := filtered[i], filtered[j]
235 if a.Key == b.Key {
236 return a.VersionID < b.VersionID
237 }
238 return a.Key < b.Key
239 })
240 return filtered, nil
241}
242func archiveBase(host string, o hostclaim.Obligation) string {
243 return "parked/" + o.Park.Incarnation + "/" + host + "/" + o.Retirement + "/"
244}
245func destination(host string, o hostclaim.Obligation, v lease.ObjectVersion) string {
246 for _, p := range prefixes(o.Park.Location) {
247 if strings.HasPrefix(v.Key, p) {
248 kind := strings.SplitN(p, "/", 2)[0]
249 return archiveBase(host, o) + kind + "/" + strings.TrimPrefix(v.Key, p) + "@" + url.QueryEscape(v.VersionID)
250 }
251 }
252 return ""
253}
254func (e Engine) current(ctx context.Context, key string) (lease.ObjectVersion, bool, error) {
255 vs, err := e.Store.ListVersions(ctx, e.Bucket, key)
256 if err != nil {
257 return lease.ObjectVersion{}, false, err
258 }
259 var got lease.ObjectVersion
260 ok := false
261 for _, v := range vs {
262 if v.Key == key && v.IsLatest {
263 if ok || v.VersionID == "" {
264 return got, false, ErrUnsafe
265 }
266 got, ok = v, true
267 }
268 }
269 if ok && got.DeleteMarker {
270 return got, false, ErrUnsafe
271 }
272 return got, ok, nil
273}
274func (e Engine) manifest(ctx context.Context, w Work, o hostclaim.Obligation) (*Manifest, error) {
275 v, ok, err := e.current(ctx, archiveBase(w.Host, o)+"manifest.json")
276 if err != nil || !ok {
277 return nil, err
278 }
279 b, err := e.Store.GetVersion(ctx, e.Bucket, v.Key, v.VersionID)
280 if err != nil {
281 return nil, err
282 }
283 var m Manifest
284 d := json.NewDecoder(bytes.NewReader(b))
285 d.DisallowUnknownFields()
286 if err := d.Decode(&m); err != nil {
287 return nil, err
288 }
289 if _, err := d.Token(); err != io.EOF {
290 return nil, ErrUnsafe
291 }
292 if m.V != 1 || m.Host != w.Host || m.Obligation != w.ID || m.Op != w.Op || m.Retirement != o.Retirement || m.Region != o.Region || m.Location != o.Park.Location || m.Incarnation != o.Park.Incarnation || m.Generation != o.Generation {
293 return nil, ErrUnsafe
294 }
295 deletes := map[Ref]lease.ObjectVersion{}
296 for _, v := range m.Delete {
297 inScope, err := scopeContains(w.Host, o.Park.Location, v.Key)
298 if err != nil || !inScope || v.VersionID == "" || isLock(w.Host, o.Park.Location, v.Key) {
299 return nil, ErrUnsafe
300 }
301 if _, ok := deletes[ref(v)]; ok {
302 return nil, ErrUnsafe
303 }
304 deletes[ref(v)] = v
305 }
306 copies := map[Ref]bool{}
307 for _, a := range m.Archive {
308 v, ok := deletes[ref(a.Source)]
309 if !ok || v.DeleteMarker || !reflect.DeepEqual(v, a.Source) || copies[ref(v)] || a.Destination.Key != destination(w.Host, o, v) || a.Destination.VersionID == "" {
310 return nil, ErrUnsafe
311 }
312 copies[ref(v)] = true
313 }
314 for r, v := range deletes {
315 if !v.DeleteMarker && !copies[r] {
316 return nil, ErrUnsafe
317 }
318 }
319 return &m, nil
320}
321func (e Engine) observe(ctx context.Context, w Work, held HeldLease, expected hostclaim.Obligation) (hostclaim.Observed, View, error) {
322 if err := held.Check(ctx); err != nil {
323 return hostclaim.Observed{}, View{}, err
324 }
325 obs, err := hostclaim.Read(ctx, e.Claims, w.Host)
326 if err != nil {
327 return obs, View{}, err
328 }
329 o, ok := find(obs.Claim, w)
330 if !ok || !reflect.DeepEqual(o, expected) {
331 return obs, View{}, ErrChanged
332 }
333 if err := e.Authorize(ctx, obs, o); err != nil {
334 return obs, View{}, err
335 }
336 v, err := e.View(ctx, obs)
337 if err != nil {
338 return obs, v, err
339 }
340 if v.ClaimETag != obs.ETag || !e.OwnsRegion(v.Region) {
341 return obs, v, ErrChanged
342 }
343 home, err := BoundHome(e.HomeRegion, obs.Claim.Op.Intent["home_region"])
344 if err != nil {
345 return obs, v, err
346 }
347 v.HomeRegion = home
348 return obs, v, nil
349}
350func (e Engine) selected(ctx context.Context, w Work, o hostclaim.Obligation, held HeldLease, vs []lease.ObjectVersion) (hostclaim.Observed, View, []lease.ObjectVersion, error) {
351 obs, v, err := e.observe(ctx, w, held, o)
352 if err != nil {
353 return obs, v, nil, err
354 }
355 live, err := protected(obs.Claim, v, o.Park.Location)
356 if err != nil {
357 return obs, v, nil, err
358 }
359 out := make([]lease.ObjectVersion, 0, len(vs))
360 for _, x := range vs {
361 inScope, err := scopeContains(w.Host, o.Park.Location, x.Key)
362 if err != nil || !inScope || x.VersionID == "" {
363 return obs, v, nil, ErrUnsafe
364 }
365 if isLock(w.Host, o.Park.Location, x.Key) {
366 return obs, v, nil, ErrUnsafe
367 }
368 if obs.Claim.Holder != nil && obs.Claim.Op.Kind != hostclaim.KindParking && o.Park.Location == obs.Claim.Holder.Location && v.sameRegion(v.Region, v.AuthoritativeRegion) {
369 if x.Key == v.Marker.Key || isBuildState(obs.Claim.Holder.Location, x.Key) {
370 if o.Kind == "park-generation" {
371 continue
372 }
373 return obs, v, nil, ErrUnsafe
374 }
375 g, ok := v.GenerationOf(ref(x))
376 if !ok {
377 return obs, v, nil, ErrUnsafe
378 }
379 g = generation(g)
380 if o.Kind == "park-generation" && g != o.Generation {
381 continue
382 }
383 if o.Kind == "park-generation" && !live[g] && v.Pinned[ref(x)] && v.FilePins[ref(x)] && strings.HasPrefix(x.Key, obs.Claim.Holder.Location+"/files/") {
384 continue
385 }
386 if live[g] || v.Pinned[ref(x)] || ref(x) == v.Marker {
387 return obs, v, nil, ErrUnsafe
388 }
389 } else if o.Kind == "park-generation" {
390 // The holder location's build state is no generation's, as the
391 // view's classifier answers for its marker: never this chore's.
392 if obs.Claim.Holder != nil && isBuildState(obs.Claim.Holder.Location, x.Key) {
393 continue
394 }
395 if v.GenerationOf == nil {
396 return obs, v, nil, ErrUnsafe
397 }
398 g, ok := v.GenerationOf(ref(x))
399 if !ok {
400 return obs, v, nil, ErrUnsafe
401 }
402 if generation(g) != o.Generation {
403 continue
404 }
405 }
406 out = append(out, x)
407 }
408 return obs, v, out, nil
409}
410
411// Run performs one resumable event. ErrPending is a scheduler retry, never an
412// acknowledgement. A newly rotated event is persisted before returning it.
413func (e Engine) Run(ctx context.Context, w Work) error { return e.run(ctx, w, nil) }
414
415// RunHeld borrows an exact location lease. The caller retains its renewal and
416// release duties through publication. Only the configured lease resolver may
417// borrow it; the same authorizer, census and engine effects still run.
418func (e Engine) RunHeld(ctx context.Context, w Work, held HeldLease) error {
419 if held == nil {
420 return ErrUnsafe
421 }
422 return e.run(ctx, w, held)
423}
424func (e Engine) run(ctx context.Context, w Work, own HeldLease) error {
425 if e.Store == nil || e.Claims == nil || e.Leases == nil || e.Now == nil || e.NewID == nil || e.View == nil || e.Authorize == nil || e.Bucket == "" || e.Region == "" {
426 return ErrUnsafe
427 }
428 initial, err := hostclaim.Read(ctx, e.Claims, w.Host)
429 if err != nil {
430 return err
431 }
432 if initial.Absent {
433 return ErrChanged
434 }
435 o, ok := find(initial.Claim, w)
436 if !ok {
437 return nil
438 }
439 if err := validateScope(w.Host, o); err != nil {
440 return err
441 }
442 if !e.OwnsRegion(o.Region) {
443 return ErrUnsafe
444 }
445 var held HeldLease
446 if own == nil {
447 held, err = e.Leases.Acquire(ctx, e.Bucket, o.Park.Location)
448 } else {
449 resolver, ok := e.Leases.(interface {
450 AcquireHeld(context.Context, string, string, HeldLease) (HeldLease, error)
451 })
452 if !ok {
453 return ErrUnsafe
454 }
455 held, err = resolver.AcquireHeld(ctx, e.Bucket, o.Park.Location, own)
456 }
457 if err != nil {
458 return err
459 }
460 if held == nil {
461 return ErrUnsafe
462 }
463 defer held.Close()
464 if bound, ok := held.(interface {
465 Context(context.Context) context.Context
466 }); ok {
467 ctx = bound.Context(ctx)
468 }
469 if _, _, err = e.observe(ctx, w, held, o); err != nil {
470 return err
471 }
472 m, err := e.manifest(ctx, w, o)
473 if err != nil {
474 return err
475 }
476 if m == nil {
477 vs, err := e.list(ctx, w.Host, o.Park.Location)
478 if err != nil {
479 return err
480 }
481 _, _, vs, err = e.selected(ctx, w, o, held, vs)
482 if err != nil {
483 return err
484 }
485 m = &Manifest{V: 1, Host: w.Host, Obligation: w.ID, Op: w.Op, Retirement: o.Retirement, Region: o.Region, Location: o.Park.Location, Incarnation: o.Park.Incarnation, Generation: o.Generation, Delete: vs}
486 for _, v := range vs {
487 if v.DeleteMarker {
488 continue
489 }
490 if _, _, _, err := e.selected(ctx, w, o, held, []lease.ObjectVersion{v}); err != nil {
491 return err
492 }
493 dst := destination(w.Host, o, v)
494 id, _, err := e.Store.CopyVersion(ctx, e.Bucket, v.Key, v.VersionID, e.Bucket, dst, true)
495 if errors.Is(err, lease.ErrDestinationExists) {
496 got, ok, getErr := e.current(ctx, dst)
497 if getErr != nil {
498 return getErr
499 }
500 if !ok || got.Size != v.Size {
501 return ErrUnsafe
502 }
503 id, err = got.VersionID, nil
504 }
505 if err != nil {
506 return err
507 }
508 if id == "" {
509 return ErrUnsafe
510 }
511 m.Archive = append(m.Archive, Copy{v, Ref{dst, id}})
512 if err := e.after("copy"); err != nil {
513 return err
514 }
515 }
516 if _, _, _, err = e.selected(ctx, w, o, held, m.Delete); err != nil {
517 return err
518 }
519 raw, err := json.Marshal(m)
520 if err != nil {
521 return err
522 }
523 _, err = e.Store.PutIfAbsent(ctx, e.Bucket, archiveBase(w.Host, o)+"manifest.json", raw)
524 if errors.Is(err, lease.ErrExists) {
525 m, err = e.manifest(ctx, w, o)
526 }
527 if err != nil {
528 return err
529 }
530 if m == nil {
531 return ErrUnsafe
532 }
533 if err := e.after("manifest"); err != nil {
534 return err
535 }
536 }
537 // Existing manifests go directly to their recorded deletes: never re-copy
538 // vanished source versions. The live-set admission check still applies.
539 for _, v := range m.Delete {
540 _, _, selected, err := e.selected(ctx, w, o, held, []lease.ObjectVersion{v})
541 if err != nil {
542 return err
543 }
544 if len(selected) != 1 {
545 return ErrUnsafe
546 }
547 if err := e.Store.DeleteVersion(ctx, e.Bucket, v.Key, v.VersionID); err != nil {
548 return err
549 }
550 if err := e.after("delete"); err != nil {
551 return err
552 }
553 }
554 started := e.Now()
555 vs, err := e.list(ctx, w.Host, o.Park.Location)
556 if err != nil {
557 return err
558 }
559 obs, view, vs, err := e.selected(ctx, w, o, held, vs)
560 if err != nil {
561 return err
562 }
563 if err := e.after("verify"); err != nil {
564 return err
565 }
566 deadline := o.Park.EvidenceAt.Add(SignatureAge + o.Park.RequestLimit)
567 if !started.After(deadline) {
568 return ErrPending
569 }
570 generations, err := parkedGenerations(obs.Claim, view, []hostclaim.Obligation{o}, m.Delete)
571 if err != nil {
572 return err
573 }
574 cover := func(c *hostclaim.Claim) error {
575 if c.Holder == nil {
576 return nil
577 }
578 if err := hostclaim.CoverParkedPredecessors(obs.Claim.Op.ID, obs.Claim.Op.Phase, o.Region, generations)(c); err != nil {
579 return err
580 }
581 if e.HomeRegion != "" && e.OwnsRegion(HomeRegion) {
582 other := HomeRegion
583 if o.Region == HomeRegion {
584 other = e.HomeRegion
585 }
586 return hostclaim.CoverParkedPredecessors(obs.Claim.Op.ID, obs.Claim.Op.Phase, other, generations)(c)
587 }
588 return nil
589 }
590 if len(vs) > 0 {
591 next := e.NewID()
592 if !token(next) || next == o.Retirement {
593 return ErrUnsafe
594 }
595 _, err = hostclaim.Update(ctx, e.Claims, obs, func(c *hostclaim.Claim) error {
596 if err := cover(c); err != nil {
597 return err
598 }
599 for i := range c.Obligations {
600 if c.Obligations[i].ID == w.ID && c.Obligations[i].Op == w.Op {
601 c.Obligations[i].Retirement = next
602 return nil
603 }
604 }
605 return ErrChanged
606 })
607 if err != nil {
608 return err
609 }
610 return ErrPending
611 }
612 if err := e.after("before-completion"); err != nil {
613 return err
614 }
615 if err := held.Check(ctx); err != nil {
616 return err
617 }
618 _, err = hostclaim.Update(ctx, e.Claims, obs, func(c *hostclaim.Claim) error {
619 if err := cover(c); err != nil {
620 return err
621 }
622 if c.QuietAfter == nil {
623 c.QuietAfter = map[string]time.Time{}
624 }
625 if c.QuietAfter[o.Region].Before(deadline) {
626 c.QuietAfter[o.Region] = deadline
627 }
628 if err := hostclaim.CompleteObligation(w.Op, w.ID)(c); err != nil {
629 return err
630 }
631 if c.Holder == nil && len(c.Obligations) == 0 && c.Op.Kind == hostclaim.KindParking && c.Op.Intent["park_sweep"] == "v1" {
632 return hostclaim.FreeAfterPark(c.Op.ID)(c)
633 }
634 return nil
635 })
636 if err != nil {
637 return err
638 }
639 if err := e.after("completion"); err != nil {
640 return err
641 }
642 return held.Release(ctx)
643}
644
645// Parked reports whether a retirement's archive is complete: its manifest
646// exists and every version in the manifest's deletion set is gone. This is
647// what the landing's parking phase waits for (moves spec §7.6); it is not the
648// acknowledgement, which still waits for the horizon and a clean listing in
649// Run. It takes no lease and writes nothing.
650//
651// Anything it cannot read, or a manifest that does not match the chore, is an
652// error and never true. A chore that is no longer in the claim returns
653// ErrChanged: it has been acknowledged (or never existed), and the caller
654// decides which from the claim it already holds.
655func (e Engine) Parked(ctx context.Context, w Work) (bool, error) {
656 if e.Store == nil || e.Claims == nil || e.Bucket == "" {
657 return false, ErrUnsafe
658 }
659 obs, err := hostclaim.Read(ctx, e.Claims, w.Host)
660 if err != nil {
661 return false, err
662 }
663 if obs.Absent {
664 return false, ErrChanged
665 }
666 o, ok := find(obs.Claim, w)
667 if !ok {
668 return false, ErrChanged
669 }
670 if o.Park == nil {
671 return false, ErrUnsafe
672 }
673 if err := validateScope(w.Host, o); err != nil {
674 return false, err
675 }
676 m, err := e.manifest(ctx, w, o)
677 if err != nil || m == nil {
678 return false, err
679 }
680 // Each distinct listed key is listed once: a version that is still there
681 // shows up under its key, and a listing that fails is an error.
682 listed := map[string]map[string]bool{}
683 var keys []string
684 for _, d := range m.Delete {
685 if listed[d.Key] == nil {
686 listed[d.Key] = map[string]bool{}
687 keys = append(keys, d.Key)
688 }
689 listed[d.Key][d.VersionID] = true
690 }
691 for _, key := range keys {
692 vs, err := e.Store.ListVersions(ctx, e.Bucket, key)
693 if err != nil {
694 return false, err
695 }
696 for _, v := range vs {
697 if v.Key != key {
698 continue
699 }
700 if v.VersionID == "" {
701 return false, ErrUnsafe
702 }
703 if listed[key][v.VersionID] {
704 return false, nil
705 }
706 }
707 }
708 return true, nil
709}
710