Fix crashes when object refreshing is performed with persistency

Calling EpochList.GetCurrentEpochState() from within the read code path
is problematic, because we might not even have an epoch. You could also
argue that refreshing doesn't necessarily need to be based on the state
of epochs, but more on the offset tracked by LocationBlobMap.
LocationBlobMap's offset could be far ahead of that of the epoch lists.

Because LocationBlobMap is unaware of a 'minimum location', invert the
logic we have right now to calculate backwards.

Author: EdSchouten

pkg/storage/object/local/block_device_backed_location_blob_map.go

@@ -11,7 +11,7 @@ type blockDeviceBackedLocationBlobMap struct {
 	sectorSizeBytes int
 	sectorCount     int64
 
-	lock         sync.Mutex
+	lock         sync.RWMutex
 	nextLocation uint64
 	sharedSector *sharedSector
 }
@@ -156,6 +156,12 @@ func (lbm *blockDeviceBackedLocationBlobMap) Put(data []byte) (uint64, error) {
 	return location, nil
 }
 
+func (lbm *blockDeviceBackedLocationBlobMap) GetNextPutLocation() uint64 {
+	lbm.lock.RLock()
+	defer lbm.lock.RUnlock()
+	return lbm.nextLocation
+}
+
 // sharedSector contains the bookkeeping of a single sector of storage
 // that stores data belonging to more than one object. Calls to
 // WriteAt() against such a sector must be synchronized, so that

pkg/storage/object/local/configuration.go

@@ -188,15 +188,15 @@ func NewStoreFromConfiguration(terminationGroup program.Group, configuration *co
 		return nil, status.Error(codes.InvalidArgument, "New region size ratio must be positive")
 	}
 	totalRatio := uint64(oldRegionSizeRatio + currentRegionSizeRatio + newRegionSizeRatio)
-	oldRegionSizeBytes := maximumLocationSpan * uint64(oldRegionSizeRatio) / totalRatio
 	currentRegionSizeBytes := maximumLocationSpan * uint64(currentRegionSizeRatio) / totalRatio
+	newRegionSizeBytes := maximumLocationSpan * uint64(newRegionSizeRatio) / totalRatio
 
 	return NewStore(
 		&globalLock,
 		referenceLocationMap,
 		locationBlobMap,
 		epochList,
-		oldRegionSizeBytes,
 		currentRegionSizeBytes,
+		newRegionSizeBytes,
 	), nil
 }

pkg/storage/object/local/in_memory_location_blob_map.go

@@ -62,3 +62,7 @@ func (lbm *inMemoryLocationBlobMap) Put(data []byte) (uint64, error) {
 	}
 	return location, nil
 }
+
+func (lbm *inMemoryLocationBlobMap) GetNextPutLocation() uint64 {
+	return lbm.nextLocation.Load()
+}

pkg/storage/object/local/location_blob_map.go

@@ -13,6 +13,7 @@ import (
 type LocationBlobMap interface {
 	Get(location uint64, sizeBytes int) ([]byte, error)
 	Put([]byte) (uint64, error)
+	GetNextPutLocation() uint64
 }
 
 // allocateLocation is a helper function for allocating a location for

pkg/storage/object/local/store.go

@@ -12,8 +12,8 @@ import (
 )
 
 type store struct {
-	oldRegionSizeBytes     uint64
 	currentRegionSizeBytes uint64
+	newRegionSizeBytes     uint64
 
 	lock                 *sync.RWMutex
 	referenceLocationMap ReferenceLocationMap
@@ -30,25 +30,24 @@ func NewStore(
 	referenceLocationMap ReferenceLocationMap,
 	locationBlobMap LocationBlobMap,
 	epochList EpochList,
-	oldRegionSizeBytes uint64,
 	currentRegionSizeBytes uint64,
+	newRegionSizeBytes uint64,
 ) object.Store[object.FlatReference, struct{}] {
 	return &store{
 		lock:                 lock,
 		referenceLocationMap: referenceLocationMap,
 		locationBlobMap:      locationBlobMap,
 		epochList:            epochList,
 
-		oldRegionSizeBytes:     oldRegionSizeBytes,
 		currentRegionSizeBytes: currentRegionSizeBytes,
+		newRegionSizeBytes:     newRegionSizeBytes,
 	}
 }
 
 func (s *store) getObjectLocation(reference object.FlatReference) (uint64, bool, error) {
 	s.lock.RLock()
-	defer s.lock.RUnlock()
-
 	location, err := s.referenceLocationMap.Get(reference, s.epochList)
+	s.lock.RUnlock()
 	if err != nil {
 		return 0, false, err
 	}
@@ -58,17 +57,16 @@ func (s *store) getObjectLocation(reference object.FlatReference) (uint64, bool,
 	// deterministic refresh threshold within the current region,
 	// derived from the object's reference. This spreads refresh
 	// operations evenly across all objects.
-	epochState, _ := s.epochList.GetCurrentEpochState()
-	distanceFromMinimum := location - epochState.MinimumLocation
+	distanceFromMaximum := int64(s.locationBlobMap.GetNextPutLocation() - location)
 
 	// Compute a deterministic threshold for this object within the
-	// current region. Objects whose distance from minimum falls
-	// below (oldRegion + threshold) need to be refreshed.
+	// current region. Objects whose distance from minimum goes
+	// above (newRegionSizeBytes + threshold) need to be refreshed.
 	// XOR with location ensures the threshold changes each time the
 	// object is relocated, preventing the same objects from always
 	// being refreshed earlier than others.
 	threshold := (location ^ binary.LittleEndian.Uint64(reference.GetRawFlatReference())) % s.currentRegionSizeBytes
-	needsRefresh := distanceFromMinimum < s.oldRegionSizeBytes+threshold
+	needsRefresh := distanceFromMaximum > int64(s.newRegionSizeBytes+threshold)
 	return location, needsRefresh, nil
 }
 

pkg/storage/object/local/store_test.go

@@ -16,7 +16,7 @@ import (
 )
 
 // newTestStore creates a store with real in-memory implementations for testing.
-func newTestStore(t *testing.T, bufferSize, oldRegionSize, currentRegionSize uint64) (object.Store[object.FlatReference, struct{}], object_local.EpochList) {
+func newTestStore(t *testing.T, bufferSize, currentRegionSizeBytes, newRegionSizeBytes uint64) (object.Store[object.FlatReference, struct{}], object_local.EpochList) {
 	var lock sync.RWMutex
 	randomNumberGenerator := random.NewFastSingleThreadedGenerator()
 	epochList := object_local.NewVolatileEpochList(bufferSize, randomNumberGenerator)
@@ -47,8 +47,8 @@ func newTestStore(t *testing.T, bufferSize, oldRegionSize, currentRegionSize uin
 		referenceLocationMap,
 		locationBlobMap,
 		epochList,
-		oldRegionSize,
-		currentRegionSize,
+		currentRegionSizeBytes,
+		newRegionSizeBytes,
 	)
 
 	return store, epochList
@@ -58,12 +58,12 @@ func TestStoreRefreshObjectInOldRegion(t *testing.T) {
 	// This test verifies that objects in the "old" region get refreshed
 	// (copied to the end of the buffer) when accessed, preventing them
 	// from being overwritten by the write cursor.
-
-	const bufferSize = 1000
-	const oldRegionSize = 200
-	const currentRegionSize = 300
-
-	store, epochList := newTestStore(t, bufferSize, oldRegionSize, currentRegionSize)
+	store, epochList := newTestStore(
+		t,
+		/* bufferSizeBytes = */ 1000,
+		/* currentRegionSize = */ 300,
+		/* newRegionSizeBytes = */ 200,
+	)
 	ctx := context.Background()
 
 	// SHA256("Hello") = 185f8db32271fe25f561a6fc938b2e264306ec304eda518007d1764826381969