Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Support backward iteration for blob feed #1314

Merged
merged 5 commits into from
Feb 25, 2025
Merged

Support backward iteration for blob feed #1314

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
ebb53f8
Support backward iteration for blob feed
jianoaix Feb 22, 2025
5bd61b1
fix
jianoaix Feb 22, 2025
b09963a
fix
jianoaix Feb 22, 2025
e3170be
until->after
jianoaix Feb 24, 2025
afcdf56
doc
jianoaix Feb 24, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
60 changes: 30 additions & 30 deletions disperser/common/v2/blobstore/dynamo_metadata_store.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -354,23 +354,23 @@ func (s *BlobMetadataStore) queryBucketBlobMetadata(
}

// GetBlobMetadataByRequestedAtForward returns blobs (as BlobMetadata) in cursor range
// (after, until] (after exclusive, until inclusive). Blobs are ordered by <RequestedAt, BlobKey>
// (after, before) (both exclusive). Blobs are retrieved and ordered by <RequestedAt, BlobKey>
// in ascending order.
//
// If limit > 0, returns at most that many blobs. If limit <= 0, returns all blobs in range.
// Also returns the cursor of the last processed blob, or nil if no blobs were processed.
func (s *BlobMetadataStore) GetBlobMetadataByRequestedAtForward(
ctx context.Context,
after BlobFeedCursor,
until BlobFeedCursor,
before BlobFeedCursor,
limit int,
) ([]*v2.BlobMetadata, *BlobFeedCursor, error) {
if !after.LessThan(&until) {
return nil, nil, errors.New("after cursor must be less than until cursor")
if !after.LessThan(&before) {
return nil, nil, errors.New("after cursor must be less than before cursor")
}
startBucket, endBucket := GetRequestedAtBucketIDRange(after.RequestedAt, until.RequestedAt)
startBucket, endBucket := GetRequestedAtBucketIDRange(after.RequestedAt, before.RequestedAt)
startKey := after.ToCursorKey()
endKey := until.ToCursorKey()
endKey := before.ToCursorKey()
result := make([]*v2.BlobMetadata, 0)
var lastProcessedCursor *BlobFeedCursor

Expand All @@ -388,8 +388,8 @@ func (s *BlobMetadataStore) GetBlobMetadataByRequestedAtForward(
if err != nil {
return nil, nil, fmt.Errorf("failed to get blob key: %w", err)
}
// Skip the after cursor's blob
if after.Equal(bm.RequestedAt, &blobKey) {
// Skip blobs at the endpoints
if after.Equal(bm.RequestedAt, &blobKey) || before.Equal(bm.RequestedAt, &blobKey) {
continue
}
result = append(result, bm)
Expand All @@ -406,22 +406,22 @@ func (s *BlobMetadataStore) GetBlobMetadataByRequestedAtForward(
}

// GetBlobMetadataByRequestedAtBackward returns blobs (as BlobMetadata) in cursor range
// [until, before) (until inclusive, before exclusive). Blobs are ordered by <RequestedAt, BlobKey>
// (after, before) (both exclusive). Blobs are retrieved and ordered by <RequestedAt, BlobKey>
// in descending order.
//
// If limit > 0, returns at most that many blobs. If limit <= 0, returns all blobs in range.
// Also returns the cursor of the last processed blob, or nil if no blobs were processed.
func (s *BlobMetadataStore) GetBlobMetadataByRequestedAtBackward(
ctx context.Context,
before BlobFeedCursor,
until BlobFeedCursor,
after BlobFeedCursor,
limit int,
) ([]*v2.BlobMetadata, *BlobFeedCursor, error) {
if !until.LessThan(&before) {
return nil, nil, errors.New("until cursor must be less than before cursor")
if !after.LessThan(&before) {
return nil, nil, errors.New("after cursor must be less than before cursor")
}
startBucket, endBucket := GetRequestedAtBucketIDRange(until.RequestedAt, before.RequestedAt)
startKey := until.ToCursorKey()
startBucket, endBucket := GetRequestedAtBucketIDRange(after.RequestedAt, before.RequestedAt)
startKey := after.ToCursorKey()
endKey := before.ToCursorKey()
result := make([]*v2.BlobMetadata, 0)
var lastProcessedCursor *BlobFeedCursor
Expand All @@ -441,8 +441,8 @@ func (s *BlobMetadataStore) GetBlobMetadataByRequestedAtBackward(
if err != nil {
return nil, nil, fmt.Errorf("failed to get blob key: %w", err)
}
// Skip the before cursor's blob
if before.Equal(bm.RequestedAt, &blobKey) {
// Skip blobs at the endpoints
if before.Equal(bm.RequestedAt, &blobKey) || after.Equal(bm.RequestedAt, &blobKey) {
continue
}
result = append(result, bm)
Expand Down Expand Up @@ -528,8 +528,8 @@ func (s *BlobMetadataStore) queryBucketAttestation(
return attestations, nil
}

// GetAttestationByAttestedAtForward returns attestations within time range (after, until]
// (after exclusive, until inclusive), ordered by AttestedAt timestamp in ascending order.
// GetAttestationByAttestedAtForward returns attestations within time range (after, before)
// (both exclusive), retrieved and ordered by AttestedAt timestamp in ascending order.
//
// The function splits the time range into buckets and queries each bucket sequentially from earliest to latest.
// Results from all buckets are combined while maintaining the ordering.
Expand All @@ -539,13 +539,13 @@ func (s *BlobMetadataStore) queryBucketAttestation(
func (s *BlobMetadataStore) GetAttestationByAttestedAtForward(
ctx context.Context,
after uint64,
until uint64,
before uint64,
limit int,
) ([]*corev2.Attestation, error) {
if after >= until {
return nil, errors.New("after must be less than until")
if after+1 > before-1 {
return nil, fmt.Errorf("no time point in exclusive time range (%d, %d)", after, before)
}
startBucket, endBucket := GetAttestedAtBucketIDRange(after, until)
startBucket, endBucket := GetAttestedAtBucketIDRange(after, before)
result := make([]*corev2.Attestation, 0)

// Traverse buckets in forward order
Expand All @@ -558,7 +558,7 @@ func (s *BlobMetadataStore) GetAttestationByAttestedAtForward(
remaining = limit - len(result)
}
// Query bucket in ascending order
bucketAttestation, err := s.queryBucketAttestation(ctx, bucket, after+1, until, remaining, true)
bucketAttestation, err := s.queryBucketAttestation(ctx, bucket, after+1, before-1, remaining, true)
if err != nil {
return nil, err
}
Expand All @@ -572,8 +572,8 @@ func (s *BlobMetadataStore) GetAttestationByAttestedAtForward(
return result, nil
}

// GetAttestationByAttestedAtBackward returns attestations within time range [until, before)
// (until inclusive, before exclusive), ordered by AttestedAt timestamp in descending order.
// GetAttestationByAttestedAtBackward returns attestations within time range (after, before)
// (both exclusive), retrieved and ordered by AttestedAt timestamp in descending order.
//
// The function splits the time range into buckets and queries each bucket sequentially from latest to earliest.
// Results from all buckets are combined while maintaining the ordering.
Expand All @@ -583,14 +583,14 @@ func (s *BlobMetadataStore) GetAttestationByAttestedAtForward(
func (s *BlobMetadataStore) GetAttestationByAttestedAtBackward(
ctx context.Context,
before uint64,
until uint64,
after uint64,
limit int,
) ([]*corev2.Attestation, error) {
if until >= before {
return nil, errors.New("until must be less than before")
if after+1 > before-1 {
return nil, fmt.Errorf("no time point in exclusive time range (%d, %d)", after, before)
}
// Note: we traverse buckets in reverse order for backward query
startBucket, endBucket := GetAttestedAtBucketIDRange(until, before)
startBucket, endBucket := GetAttestedAtBucketIDRange(after, before)
result := make([]*corev2.Attestation, 0)

// Traverse buckets in reverse order
Expand All @@ -603,7 +603,7 @@ func (s *BlobMetadataStore) GetAttestationByAttestedAtBackward(
remaining = limit - len(result)
}
// Query bucket in descending order
bucketAttestation, err := s.queryBucketAttestation(ctx, bucket, until, before-1, remaining, false)
bucketAttestation, err := s.queryBucketAttestation(ctx, bucket, after+1, before-1, remaining, false)
if err != nil {
return nil, err
}
Expand Down
Loading
Loading