storage

package

Go to main page

Versions in this module

v1

v1.6.0

Jun 2, 2026

Changes in this version

+ var ErrNotFound = errors.New("not found")

type MemoryS3Client

+ func (m *MemoryS3Client) DeleteIndex(ctx context.Context, key string) error

+ func (m *MemoryS3Client) DeleteSegment(ctx context.Context, key string) error

type PartitionLogConfig

+ Logger *slog.Logger

+ type RecoveredPartition struct

+ LastOffset int64

+ Partition int32

+ Segments []RecoveredSegment

+ SegmentsCopied int

+ type RecoveredSegment struct

+ BaseOffset int64

+ CreatedAt time.Time

+ LastOffset int64

+ Partition int32

+ SizeBytes int64

+ SourceIndex string

+ SourceKey string

+ TargetIndex string

+ TargetKey string

type S3Client

+ DeleteIndex func(ctx context.Context, key string) error

+ DeleteSegment func(ctx context.Context, key string) error

type S3Config

+ MaxConnections int

+ type TopicRecoveryConfig struct

+ Partitions []int32

+ RestoreTo time.Time

+ SourceNamespace string

+ SourceTopic string

+ TargetNamespace string

+ TargetTopic string

+ type TopicRecoveryResult struct

+ Partitions []RecoveredPartition

+ RestoreTo time.Time

+ SegmentsCopied int

+ SourceNamespace string

+ SourceTopic string

+ TargetNamespace string

+ TargetTopic string

+ func RecoverTopicToTimestamp(ctx context.Context, s3 S3Client, cfg TopicRecoveryConfig) (*TopicRecoveryResult, error)

v1.5.0

Jan 30, 2026

v1.4.2

Jan 19, 2026

Changes in this version

+ var ErrOffsetOutOfRange = errors.New("offset out of range")

+ func CountRecordBatchMessages(recordSet []byte) int

+ func PatchRecordBatchBaseOffset(batch *RecordBatch, baseOffset int64)

+ type AppendResult struct

+ BaseOffset int64

+ LastOffset int64

+ type ByteRange struct

+ End int64

+ Start int64

+ type IndexBuilder struct

+ func NewIndexBuilder(interval int32) *IndexBuilder

+ func (b *IndexBuilder) BuildBytes() ([]byte, error)

+ func (b *IndexBuilder) Entries() []*IndexEntry

+ func (b *IndexBuilder) MaybeAdd(offset int64, position int32, batchMessages int32)

+ type IndexEntry struct

+ Offset int64

+ Position int32

+ func ParseIndex(data []byte) ([]*IndexEntry, error)

+ type MemoryS3Client struct

+ func NewMemoryS3Client() *MemoryS3Client

+ func (m *MemoryS3Client) DownloadIndex(ctx context.Context, key string) ([]byte, error)

+ func (m *MemoryS3Client) DownloadSegment(ctx context.Context, key string, rng *ByteRange) ([]byte, error)

+ func (m *MemoryS3Client) EnsureBucket(ctx context.Context) error

+ func (m *MemoryS3Client) ListSegments(ctx context.Context, prefix string) ([]S3Object, error)

+ func (m *MemoryS3Client) UploadIndex(ctx context.Context, key string, body []byte) error

+ func (m *MemoryS3Client) UploadSegment(ctx context.Context, key string, body []byte) error

+ type PartitionLog struct

+ func NewPartitionLog(namespace string, topic string, partition int32, startOffset int64, ...) *PartitionLog

+ func (l *PartitionLog) AppendBatch(ctx context.Context, batch RecordBatch) (*AppendResult, error)

+ func (l *PartitionLog) EarliestOffset() int64

+ func (l *PartitionLog) Flush(ctx context.Context) error

+ func (l *PartitionLog) Read(ctx context.Context, offset int64, maxBytes int32) ([]byte, error)

+ func (l *PartitionLog) RestoreFromS3(ctx context.Context) (int64, error)

+ type PartitionLogConfig struct

+ Buffer WriteBufferConfig

+ CacheEnabled bool

+ ReadAheadSegments int

+ Segment SegmentWriterConfig

+ type RecordBatch struct

+ BaseOffset int64

+ Bytes []byte

+ LastOffsetDelta int32

+ MessageCount int32

+ func NewRecordBatchFromBytes(data []byte) (RecordBatch, error)

+ type S3Client interface

+ DownloadIndex func(ctx context.Context, key string) ([]byte, error)

+ DownloadSegment func(ctx context.Context, key string, rng *ByteRange) ([]byte, error)

+ EnsureBucket func(ctx context.Context) error

+ ListSegments func(ctx context.Context, prefix string) ([]S3Object, error)

+ UploadIndex func(ctx context.Context, key string, body []byte) error

+ UploadSegment func(ctx context.Context, key string, body []byte) error

+ func NewS3Client(ctx context.Context, cfg S3Config) (S3Client, error)

+ type S3Config struct

+ AccessKeyID string

+ Bucket string

+ Endpoint string

+ ForcePathStyle bool

+ KMSKeyARN string

+ Region string

+ SecretAccessKey string

+ SessionToken string

+ type S3Object struct

+ Key string

+ Size int64

+ type SegmentArtifact struct

+ BaseOffset int64

+ CreatedAt time.Time

+ IndexBytes []byte

+ LastOffset int64

+ MessageCount int32

+ RelativeIndex []*IndexEntry

+ SegmentBytes []byte

+ func BuildSegment(cfg SegmentWriterConfig, batches []RecordBatch, created time.Time) (*SegmentArtifact, error)

+ type SegmentWriterConfig struct

+ IndexIntervalMessages int32

+ type WriteBuffer struct

+ func NewWriteBuffer(cfg WriteBufferConfig) *WriteBuffer

+ func (b *WriteBuffer) Append(batch RecordBatch)

+ func (b *WriteBuffer) Drain() []RecordBatch

+ func (b *WriteBuffer) ShouldFlush(now time.Time) bool

+ func (b *WriteBuffer) Size() int

+ type WriteBufferConfig struct

+ FlushInterval time.Duration

+ MaxBatches int

+ MaxBytes int

+ MaxMessages int

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL