amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1// Package leasetest holds test doubles for internal/lease. VersionedBucket
2// is the reference model of a versioned S3 bucket that every plan of the
3// host lifecycle claim tests against (spec §13).
4package leasetest
5
6import (
7 "bytes"
8 "context"
9 "crypto/md5"
10 "encoding/hex"
11 "fmt"
12 "io"
13 "sort"
14 "strings"
15 "sync"
16 "time"
17
18 "github.com/carlosframework/platform/internal/lease"
19)
20
21type version struct {
22 id string
23 body []byte
24 etag string
25 deleteMarker bool
26 modified time.Time
27}
28
29// VersionedBucket is an in-memory, versioned, multi-bucket S3 model.
30type VersionedBucket struct {
31 mu sync.Mutex
32 buckets map[string]map[string][]version // bucket -> key -> versions, newest first
33 next int
34 clock time.Time
35
36 PageSize int // listing page size; default 1000
37 FailListAfterPages int // fail a listing once this many pages have been served; 0 never
38 FailOn func(op, key string) error // op: list|get|copy|delete|put
39}
40
41func NewVersionedBucket() *VersionedBucket {
42 return &VersionedBucket{buckets: map[string]map[string][]version{}, PageSize: 1000,
43 clock: time.Date(2026, 10, 5, 0, 0, 0, 0, time.UTC)}
44}
45
46func md5hex(b []byte) string { s := md5.Sum(b); return hex.EncodeToString(s[:]) }
47
48func (f *VersionedBucket) AddBucket(name string) {
49 f.mu.Lock()
50 defer f.mu.Unlock()
51 if f.buckets[name] == nil {
52 f.buckets[name] = map[string][]version{}
53 }
54}
55
56// --- locked helpers: callers hold f.mu ---
57
58func (f *VersionedBucket) fail(op, key string) error {
59 if f.FailOn != nil {
60 return f.FailOn(op, key)
61 }
62 return nil
63}
64
65func (f *VersionedBucket) known(bucket string) error {
66 if _, ok := f.buckets[bucket]; !ok {
67 return fmt.Errorf("leasetest: no bucket %q", bucket)
68 }
69 return nil
70}
71
72func (f *VersionedBucket) push(bucket, key string, v version) string {
73 f.next++
74 f.clock = f.clock.Add(time.Second)
75 v.id, v.modified = fmt.Sprintf("v%06d", f.next), f.clock
76 f.buckets[bucket][key] = append([]version{v}, f.buckets[bucket][key]...)
77 return v.id
78}
79
80func (f *VersionedBucket) hasCurrent(bucket, key string) bool {
81 vs := f.buckets[bucket][key]
82 return len(vs) > 0 && !vs[0].deleteMarker
83}
84
85func (f *VersionedBucket) find(bucket, key, id string) (version, bool) {
86 for _, v := range f.buckets[bucket][key] {
87 if v.id == id {
88 return v, true
89 }
90 }
91 return version{}, false
92}
93
94// listAll is the whole ordered listing under prefix (key order, newest
95// first within a key).
96func (f *VersionedBucket) listAll(bucket, prefix string) []lease.ObjectVersion {
97 o := f.buckets[bucket]
98 keys := make([]string, 0, len(o))
99 for k := range o {
100 if strings.HasPrefix(k, prefix) {
101 keys = append(keys, k)
102 }
103 }
104 sort.Strings(keys)
105 var all []lease.ObjectVersion
106 for _, k := range keys {
107 for i, v := range o[k] {
108 ov := lease.ObjectVersion{Key: k, VersionID: v.id, IsLatest: i == 0, DeleteMarker: v.deleteMarker, LastModified: v.modified}
109 if !v.deleteMarker {
110 ov.ETag, ov.Size = v.etag, int64(len(v.body))
111 }
112 all = append(all, ov)
113 }
114 }
115 return all
116}
117
118// pageOf returns page number page (0-based) of all, whether more follow,
119// and the failure FailListAfterPages injects.
120func (f *VersionedBucket) pageOf(all []lease.ObjectVersion, page int) ([]lease.ObjectVersion, bool, error) {
121 if f.FailListAfterPages > 0 && page >= f.FailListAfterPages {
122 return nil, false, fmt.Errorf("leasetest: injected listing failure on page %d", page+1)
123 }
124 size := f.PageSize
125 if size <= 0 {
126 size = 1000
127 }
128 start := page * size
129 if start > len(all) {
130 start = len(all)
131 }
132 end := start + size
133 if end >= len(all) {
134 return all[start:], false, nil
135 }
136 return all[start:end], true, nil
137}
138
139// --- exported helpers for tests ---
140
141func (f *VersionedBucket) Put(bucket, key string, body []byte) string {
142 f.mu.Lock()
143 defer f.mu.Unlock()
144 return f.push(bucket, key, version{body: append([]byte(nil), body...), etag: md5hex(body)})
145}
146
147func (f *VersionedBucket) PutMultipart(bucket, key string, body []byte) string {
148 f.mu.Lock()
149 defer f.mu.Unlock()
150 return f.push(bucket, key, version{body: append([]byte(nil), body...), etag: md5hex(body) + "-2"})
151}
152
153func (f *VersionedBucket) BareDelete(bucket, key string) string {
154 f.mu.Lock()
155 defer f.mu.Unlock()
156 return f.push(bucket, key, version{deleteMarker: true})
157}
158
159func (f *VersionedBucket) LatestID(bucket, key string) string {
160 f.mu.Lock()
161 defer f.mu.Unlock()
162 if vs := f.buckets[bucket][key]; len(vs) > 0 {
163 return vs[0].id
164 }
165 return ""
166}
167
168func (f *VersionedBucket) Current(bucket, key string) ([]byte, bool) {
169 f.mu.Lock()
170 defer f.mu.Unlock()
171 vs := f.buckets[bucket][key]
172 if len(vs) == 0 || vs[0].deleteMarker {
173 return nil, false
174 }
175 return vs[0].body, true
176}
177
178// Snapshot renders the whole state, for before/after comparison.
179func (f *VersionedBucket) Snapshot() string {
180 f.mu.Lock()
181 defer f.mu.Unlock()
182 var b strings.Builder
183 names := make([]string, 0, len(f.buckets))
184 for n := range f.buckets {
185 names = append(names, n)
186 }
187 sort.Strings(names)
188 for _, n := range names {
189 for _, v := range f.listAll(n, "") {
190 fmt.Fprintf(&b, "%s/%s@%s:%v:%s\n", n, v.Key, v.VersionID, v.DeleteMarker, v.ETag)
191 }
192 }
193 return b.String()
194}
195
196// --- lease.VersionedStore ---
197
198func (f *VersionedBucket) ListVersions(ctx context.Context, bucket, prefix string) ([]lease.ObjectVersion, error) {
199 f.mu.Lock()
200 defer f.mu.Unlock()
201 if err := f.known(bucket); err != nil {
202 return nil, err
203 }
204 if err := f.fail("list", prefix); err != nil {
205 return nil, err
206 }
207 all := f.listAll(bucket, prefix)
208 var out []lease.ObjectVersion
209 for page := 0; ; page++ {
210 vs, more, err := f.pageOf(all, page)
211 if err != nil {
212 return nil, err
213 }
214 out = append(out, vs...)
215 if !more {
216 return out, nil
217 }
218 }
219}
220
221func (f *VersionedBucket) read(bucket, key, versionID string) ([]byte, error) {
222 if versionID == "" {
223 return nil, lease.ErrNoVersionID
224 }
225 if err := f.known(bucket); err != nil {
226 return nil, err
227 }
228 if err := f.fail("get", key); err != nil {
229 return nil, err
230 }
231 v, ok := f.find(bucket, key, versionID)
232 if !ok {
233 return nil, lease.ErrMirrorNotFound
234 }
235 if v.deleteMarker {
236 return nil, lease.ErrIsDeleteMarker
237 }
238 return append([]byte(nil), v.body...), nil
239}
240
241func (f *VersionedBucket) GetVersion(ctx context.Context, bucket, key, versionID string) ([]byte, error) {
242 f.mu.Lock()
243 defer f.mu.Unlock()
244 return f.read(bucket, key, versionID)
245}
246
247func (f *VersionedBucket) OpenVersion(ctx context.Context, bucket, key, versionID string) (io.ReadCloser, error) {
248 f.mu.Lock()
249 defer f.mu.Unlock()
250 b, err := f.read(bucket, key, versionID)
251 if err != nil {
252 return nil, err
253 }
254 return io.NopCloser(bytes.NewReader(b)), nil
255}
256
257func versionRange(body []byte, offset, length int64) ([]byte, error) {
258 if offset < 0 || length == 0 || (length > 0 && offset > (1<<63-1)-(length-1)) || offset >= int64(len(body)) {
259 return nil, fmt.Errorf("leasetest: invalid version range")
260 }
261 n := int64(len(body)) - offset
262 if length > 0 && length < n {
263 n = length
264 }
265 return body[offset : offset+n], nil
266}
267
268func (f *VersionedBucket) OpenVersionRange(ctx context.Context, bucket, key, versionID string, offset, length int64) (io.ReadCloser, error) {
269 f.mu.Lock()
270 defer f.mu.Unlock()
271 b, err := f.read(bucket, key, versionID)
272 if err != nil {
273 return nil, err
274 }
275 b, err = versionRange(b, offset, length)
276 if err != nil {
277 return nil, err
278 }
279 return io.NopCloser(bytes.NewReader(b)), nil
280}
281
282func (f *VersionedBucket) copyLocked(srcBucket, srcKey, srcVersionID, dstBucket, dstKey string, createOnly bool) (string, string, error) {
283 if srcVersionID == "" {
284 return "", "", lease.ErrNoVersionID
285 }
286 if err := f.known(srcBucket); err != nil {
287 return "", "", err
288 }
289 if err := f.known(dstBucket); err != nil {
290 return "", "", err
291 }
292 if err := f.fail("copy", srcKey); err != nil {
293 return "", "", err
294 }
295 v, ok := f.find(srcBucket, srcKey, srcVersionID)
296 if !ok {
297 return "", "", lease.ErrMirrorNotFound
298 }
299 if v.deleteMarker {
300 return "", "", lease.ErrIsDeleteMarker
301 }
302 if createOnly && f.hasCurrent(dstBucket, dstKey) {
303 return "", "", lease.ErrDestinationExists
304 }
305 etag := md5hex(v.body) // a copy is single-part (§13 row 9)
306 return f.push(dstBucket, dstKey, version{body: append([]byte(nil), v.body...), etag: etag}), etag, nil
307}
308
309func (f *VersionedBucket) CopyVersion(ctx context.Context, srcBucket, srcKey, srcVersionID, dstBucket, dstKey string, createOnly bool) (string, string, error) {
310 f.mu.Lock()
311 defer f.mu.Unlock()
312 return f.copyLocked(srcBucket, srcKey, srcVersionID, dstBucket, dstKey, createOnly)
313}
314
315func (f *VersionedBucket) deleteLocked(bucket, key, versionID string) error {
316 if versionID == "" {
317 return lease.ErrNoVersionID
318 }
319 if err := f.known(bucket); err != nil {
320 return err
321 }
322 if err := f.fail("delete", key); err != nil {
323 return err
324 }
325 vs := f.buckets[bucket][key]
326 for i, v := range vs {
327 if v.id == versionID {
328 rest := append(vs[:i:i], vs[i+1:]...)
329 if len(rest) == 0 {
330 delete(f.buckets[bucket], key)
331 } else {
332 f.buckets[bucket][key] = rest
333 }
334 return nil
335 }
336 }
337 return nil
338}
339
340func (f *VersionedBucket) DeleteVersion(ctx context.Context, bucket, key, versionID string) error {
341 f.mu.Lock()
342 defer f.mu.Unlock()
343 return f.deleteLocked(bucket, key, versionID)
344}
345
346func (f *VersionedBucket) putIfAbsentLocked(bucket, key string, body []byte) (string, error) {
347 if err := f.known(bucket); err != nil {
348 return "", err
349 }
350 if err := f.fail("put", key); err != nil {
351 return "", err
352 }
353 if f.hasCurrent(bucket, key) {
354 return "", lease.ErrExists
355 }
356 return f.push(bucket, key, version{body: append([]byte(nil), body...), etag: md5hex(body)}), nil
357}
358
359// putLocked is a plain PUT: it stores a new version beside the existing
360// ones, as S3 with versioning on does. putIfAbsentLocked is the create-only
361// variant (If-None-Match: *).
362func (f *VersionedBucket) putLocked(bucket, key string, body []byte) (string, error) {
363 if err := f.known(bucket); err != nil {
364 return "", err
365 }
366 if err := f.fail("put", key); err != nil {
367 return "", err
368 }
369 return f.push(bucket, key, version{body: append([]byte(nil), body...), etag: md5hex(body)}), nil
370}
371
372// bareDeleteLocked is a DELETE naming no version: S3 with versioning on
373// adds a delete marker and removes nothing.
374func (f *VersionedBucket) bareDeleteLocked(bucket, key string) (string, error) {
375 if err := f.known(bucket); err != nil {
376 return "", err
377 }
378 if err := f.fail("delete", key); err != nil {
379 return "", err
380 }
381 return f.push(bucket, key, version{deleteMarker: true}), nil
382}
383
384func (f *VersionedBucket) PutIfAbsent(ctx context.Context, bucket, key string, body []byte) (string, error) {
385 f.mu.Lock()
386 defer f.mu.Unlock()
387 return f.putIfAbsentLocked(bucket, key, body)
388}
389
390var _ lease.VersionedStore = (*VersionedBucket)(nil)
391
392var _ lease.VersionRangeOpener = (*VersionedBucket)(nil)
393