s3store

package
v0.1.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 10, 2026 License: Apache-2.0 Imports: 17 Imported by: 0

Documentation

Overview

Package s3store is the storage engine behind the doze-aws s3 service: bucket and object-version metadata in bbolt, object bodies as one file per version on disk. Bodies are only ever streamed (temp-file + rename in, ReadSeeker out) — steady-state memory is independent of object and bucket size, the same flat-memory invariant doze-kafka's log store keeps.

bbolt layout: 

buckets              name        -> Bucket JSON
ver:<bucket>         objKey \x00 ^seq (big-endian) -> ObjectVersion JSON (newest first per key)
cur:<bucket>         objKey      -> ver: key of the latest version (absent when latest is a delete marker)
up:<bucket>          uploadID    -> Upload JSON
parts:<bucket>       uploadID \x00 be32(part) -> Part JSON

Blob files live at blobs/<bucket>/<shard>/<blobID>; temp writes under tmp/.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func ErrNoSuchBucket 

func ErrNoSuchBucket(name string) *awshttp.APIError

func ErrNoSuchKey 

func ErrNoSuchKey(key string) *awshttp.APIError

Types

type Bucket 

type Bucket struct {
	Name       string            `json:"name"`
	Created    int64             `json:"created"`              // unix seconds
	Versioning string            `json:"versioning,omitempty"` // "" | Enabled | Suspended
	Tags       map[string]string `json:"tags,omitempty"`

	// Functional configs (evaluated locally).
	CORS       string `json:"cors,omitempty"`        // CORS XML document
	Lifecycle  string `json:"lifecycle,omitempty"`   // lifecycle XML document
	Website    string `json:"website,omitempty"`     // website XML document
	ObjectLock string `json:"object_lock,omitempty"` // object-lock XML document

	// Round-trip-only configs.
	Policy       string `json:"policy,omitempty"`
	ACL          string `json:"acl,omitempty"` // canned ACL name
	Encryption   string `json:"encryption,omitempty"`
	Logging      string `json:"logging,omitempty"`
	Notification string `json:"notification,omitempty"`
	Replication  string `json:"replication,omitempty"`
	Accelerate   string `json:"accelerate,omitempty"`
	RequestPays  string `json:"request_pays,omitempty"`
}

Bucket is a bucket's durable definition. Config documents with no local behavior are stored raw and returned faithfully (Tier C).

type CompletedPart 

type CompletedPart struct {
	Number      int
	ETag        string
	ChecksumVal string // optional client-declared checksum
}

CompletedPart is the client's view of one part in CompleteMultipartUpload.

type ListEntry 

type ListEntry struct {
	Key          string
	Size         int64
	ETag         string
	LastModified int64
	StorageClass string
	ChecksumAlg  string
}

ListEntry is one object in a list result.

type ListResult 

type ListResult struct {
	Entries        []ListEntry
	CommonPrefixes []string
	IsTruncated    bool
	NextToken      string // next StartAfter/Marker/ContinuationToken value (a key)
}

ListResult is a paged ListObjects(V2) result.

type ObjectVersion 

type ObjectVersion struct {
	Key          string            `json:"key"`
	VersionID    string            `json:"version_id"` // "null" on unversioned buckets
	Blob         string            `json:"blob,omitempty"`
	Size         int64             `json:"size"`
	ETag         string            `json:"etag"` // without quotes
	ChecksumAlg  string            `json:"checksum_alg,omitempty"`
	ChecksumVal  string            `json:"checksum_val,omitempty"`  // base64 (composite form for multipart)
	ChecksumType string            `json:"checksum_type,omitempty"` // FULL_OBJECT | COMPOSITE
	ContentType  string            `json:"content_type,omitempty"`
	Meta         map[string]string `json:"meta,omitempty"`    // x-amz-meta-*
	Headers      map[string]string `json:"headers,omitempty"` // Cache-Control, Content-Disposition, ...
	Tags         map[string]string `json:"tags,omitempty"`
	StorageClass string            `json:"storage_class,omitempty"`
	DeleteMarker bool              `json:"delete_marker,omitempty"`
	LastModified int64             `json:"last_modified"` // unix seconds

	// Object lock (enforced).
	RetainUntil int64  `json:"retain_until,omitempty"` // unix seconds
	RetainMode  string `json:"retain_mode,omitempty"`  // GOVERNANCE | COMPLIANCE
	LegalHold   bool   `json:"legal_hold,omitempty"`
	// contains filtered or unexported fields
}

ObjectVersion is one stored version of one object key.

func (*ObjectVersion) SeqKey 

func (v *ObjectVersion) SeqKey() []byte

SeqKey returns the bbolt key the version was loaded from.

type Part 

type Part struct {
	Number       int    `json:"number"`
	Blob         string `json:"blob"`
	Size         int64  `json:"size"`
	ETag         string `json:"etag"`
	ChecksumVal  string `json:"checksum_val,omitempty"`
	LastModified int64  `json:"last_modified"`
}

Part is one uploaded part.

type Store 

type Store struct {
	Logf func(format string, args ...any)
	// contains filtered or unexported fields
}

Store is the S3 storage engine.

func Open 

func Open(dataDir string) (*Store, error)

Open opens (or initializes) the store under dataDir.

func (*Store) AbortUpload 

func (s *Store) AbortUpload(bucket, uploadID string) error

AbortUpload discards an upload and its part blobs.

func (*Store) Close 

func (s *Store) Close() error

Close closes the metadata database.

func (*Store) CompleteUpload 

func (s *Store) CompleteUpload(bucket, uploadID string, declared []CompletedPart, ifNoneMatch, ifMatch string) (*ObjectVersion, error)

CompleteUpload validates the part list, assembles the final blob, computes the multipart ETag ("md5-of-md5s-N") and composite/full checksum, commits the object version, and cleans up the upload. checksumFull computes the FULL_OBJECT checksum from the assembled blob when the algorithm needs it.

func (*Store) ConcatBlobs 

func (s *Store) ConcatBlobs(bucket string, blobs []string) (blob string, size int64, err error)

ConcatBlobs streams the given blobs, in order, into one new blob — CompleteMultipartUpload's assembly step. Constant memory.

func (*Store) CreateBucket 

func (s *Store) CreateBucket(name string, objectLock bool) error

CreateBucket creates a bucket; recreating an existing one succeeds (AWS returns 200 for the owner in us-east-1, and local buckets are always yours).

func (*Store) CreateUpload 

func (s *Store) CreateUpload(bucket string, up Upload) (*Upload, error)

CreateUpload starts a multipart upload.

func (*Store) DeleteBlob 

func (s *Store) DeleteBlob(blob string)

DeleteBlob removes a blob file; missing files are fine (idempotent).

func (*Store) DeleteBucket 

func (s *Store) DeleteBucket(name string) error

DeleteBucket removes an empty bucket.

func (*Store) DeleteObject 

func (s *Store) DeleteObject(bucket, key, versionID string, bypassGovernance bool) (bool, string, error)

DeleteObject implements DELETE semantics:

  • versionID given: remove that exact version (and its blob);
  • no versionID, versioning Enabled: insert a delete marker;
  • no versionID, otherwise: remove the "null"/current version.

Returns (deleteMarkerCreated, versionIDAffected).

func (*Store) GetBucket 

func (s *Store) GetBucket(name string) (*Bucket, error)

GetBucket loads a bucket definition.

func (*Store) GetUpload 

func (s *Store) GetUpload(bucket, uploadID string) (*Upload, error)

GetUpload loads an in-progress upload.

func (*Store) GetVersion 

func (s *Store) GetVersion(bucket, key, versionID string) (*ObjectVersion, error)

GetVersion fetches a key's current version, or a specific one. Delete markers surface as (version, MethodNotAllowed-flavored 404) the way S3 reports them.

func (*Store) ListBuckets 

func (s *Store) ListBuckets() ([]Bucket, error)

ListBuckets returns every bucket, sorted by name (bbolt order).

func (*Store) ListObjects 

func (s *Store) ListObjects(bucket, prefix, delimiter, after string, max int) (*ListResult, error)

ListObjects walks the current (visible) objects of a bucket with prefix/delimiter/paging semantics shared by ListObjects and ListObjectsV2. after is the exclusive start key (Marker / StartAfter / decoded token).

func (*Store) ListVersions 

func (s *Store) ListVersions(bucket, prefix, delimiter, keyMarker, versionMarker string, max int) (*VersionsResult, error)

ListVersions walks every version (delete markers included), newest-first within each key.

func (*Store) OpenBlob 

func (s *Store) OpenBlob(blob string) (*os.File, error)

OpenBlob opens a blob for streaming reads (supports ranges via Seek).

func (*Store) Parts 

func (s *Store) Parts(bucket, uploadID string) ([]Part, error)

Parts lists an upload's parts in part-number order.

func (*Store) PutPart 

func (s *Store) PutPart(bucket, uploadID string, part Part) error

PutPart records an uploaded part (blob already written). Re-uploading a part number replaces it.

func (*Store) PutVersion 

func (s *Store) PutVersion(bucket string, v ObjectVersion, ifNoneMatch, ifMatch string) (*ObjectVersion, error)

PutVersion commits a fully-written blob as the new version of a key, honoring the bucket's versioning state. Returns the stored version.

Conditional writes: ifNoneMatch ("*" — fail if the key exists) and ifMatch (fail unless the current ETag matches) implement S3's conditional PUT.

func (*Store) SetClock 

func (s *Store) SetClock(fn func() time.Time)

SetClock overrides the clock (tests).

func (*Store) UpdateBucket 

func (s *Store) UpdateBucket(name string, fn func(*Bucket) error) error

UpdateBucket applies fn to a bucket and persists it.

func (*Store) UpdateVersion 

func (s *Store) UpdateVersion(bucket string, v *ObjectVersion, fn func(*ObjectVersion) error) error

UpdateVersion persists a mutation (tagging, lock fields) to an existing version record.

func (*Store) Uploads 

func (s *Store) Uploads(bucket, prefix string) ([]Upload, error)

Uploads lists in-progress uploads, sorted by key then id.

func (*Store) WriteBlob 

func (s *Store) WriteBlob(bucket string, r io.Reader) (blob string, size int64, err error)

WriteBlob streams r into a new blob file for the bucket (temp file + fsync + rename, so a crash never leaves a partial blob visible) and returns the blob's store-relative path and size.

type Upload 

type Upload struct {
	ID          string            `json:"id"`
	Key         string            `json:"key"`
	Initiated   int64             `json:"initiated"`
	ContentType string            `json:"content_type,omitempty"`
	Meta        map[string]string `json:"meta,omitempty"`
	Headers     map[string]string `json:"headers,omitempty"`
	Tags        map[string]string `json:"tags,omitempty"`
	ChecksumAlg string            `json:"checksum_alg,omitempty"`
	// ChecksumType is FULL_OBJECT or COMPOSITE (how the final checksum is built).
	ChecksumType string `json:"checksum_type,omitempty"`
}

Upload is one in-progress multipart upload.

type VersionEntry 

type VersionEntry struct {
	ObjectVersion
	IsLatest bool
}

VersionEntry is one row of a ListObjectVersions result.

type VersionsResult 

type VersionsResult struct {
	Versions        []VersionEntry
	CommonPrefixes  []string
	IsTruncated     bool
	NextKeyMarker   string
	NextVersionMark string
}

VersionsResult is a paged ListObjectVersions result.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.