diff --git a/component/attr_cache/attr_cache_test.go b/component/attr_cache/attr_cache_test.go
index b39b347e6..17b699d80 100644
--- a/component/attr_cache/attr_cache_test.go
+++ b/component/attr_cache/attr_cache_test.go
@@ -1420,6 +1420,42 @@ func (suite *attrCacheTestSuite) TestTruncateFile() {
 	suite.assert.True(checkItem.exists())
 }
 
+// Attributes returned by GetAttr are read by callers after the cache lock is released,
+// so later cache updates must not modify them in place
+func (suite *attrCacheTestSuite) TestGetAttrResultNotModifiedByCacheUpdates() {
+	defer suite.cleanupTest()
+	path := "a"
+	suite.addPathToCache(path)
+
+	attr, err := suite.attrCache.GetAttr(internal.GetAttrOptions{Name: path})
+	suite.assert.NoError(err)
+	before := *attr
+
+	readerDone := make(chan struct{})
+	go func() {
+		defer close(readerDone)
+		for range 1000 {
+			_ = attr.Size
+			_ = attr.Mtime
+			_ = attr.Mode
+		}
+	}()
+
+	truncateOptions := internal.TruncateFileOptions{Name: path, NewSize: 1234}
+	suite.mock.EXPECT().TruncateFile(truncateOptions).Return(nil)
+	suite.assert.NoError(suite.attrCache.TruncateFile(truncateOptions))
+	chmodOptions := internal.ChmodOptions{Name: path, Mode: 0600}
+	suite.mock.EXPECT().Chmod(chmodOptions).Return(nil)
+	suite.assert.NoError(suite.attrCache.Chmod(chmodOptions))
+	<-readerDone
+
+	suite.assert.Equal(before, *attr)
+	updated, err := suite.attrCache.GetAttr(internal.GetAttrOptions{Name: path})
+	suite.assert.NoError(err)
+	suite.assert.EqualValues(1234, updated.Size)
+	suite.assert.Equal(os.FileMode(0600), updated.Mode.Perm())
+}
+
 // Tests CopyFromFile
 func (suite *attrCacheTestSuite) TestCopyFromFileError() {
 	defer suite.cleanupTest()
diff --git a/component/attr_cache/cacheMap.go b/component/attr_cache/cacheMap.go
index 8c9f59f6c..c76c68600 100644
--- a/component/attr_cache/cacheMap.go
+++ b/component/attr_cache/cacheMap.go
@@ -265,7 +265,15 @@ func (value *attrCacheItem) markInCloud(inCloud bool) {
 	}
 }
 
+// cloneAttr replaces the item's attributes with a private copy so they can be modified.
+// Callers may still be reading the previous copy after the cache lock was released.
+func (value *attrCacheItem) cloneAttr() {
+	attr := *value.attr
+	value.attr = &attr
+}
+
 func (value *attrCacheItem) setSize(size int64, changedAt time.Time) {
+	value.cloneAttr()
 	value.attr.Mtime = changedAt
 	value.attr.Ctime = changedAt
 	value.attr.Size = size
@@ -276,6 +284,7 @@ func (value *attrCacheItem) touchModifyAndChangeTimes(changedAt time.Time) {
 	if value == nil || !value.exists() {
 		return
 	}
+	value.cloneAttr()
 	value.attr.Mtime = changedAt
 	value.attr.Ctime = changedAt
 	value.cachedAt = changedAt
@@ -287,6 +296,7 @@ func (value *attrCacheItem) setMode(mode os.FileMode) {
 		currentType = mode & os.ModeType
 	}
 	modeBits := mode & (os.ModePerm | os.ModeSetuid | os.ModeSetgid | os.ModeSticky)
+	value.cloneAttr()
 	value.attr.Mode = currentType | modeBits
 	value.attr.Flags.Clear(internal.PropFlagModeDefault)
 	value.attr.Ctime = time.Now()
diff --git a/component/azstorage/block_blob_test.go b/component/azstorage/block_blob_test.go
index 2ed5f70dd..e4f6f7b6e 100644
--- a/component/azstorage/block_blob_test.go
+++ b/component/azstorage/block_blob_test.go
@@ -356,7 +356,7 @@ func generateContainerName() string {
 
 func createTestContainerWithRetry(create func() error) error {
 	var err error
-	for i := 0; i < 5; i++ {
+	for i := range 5 {
 		err = create()
 		if err == nil {
 			return nil
diff --git a/component/block_cache/block_cache.go b/component/block_cache/block_cache.go
index 071c1cb44..6f2ff52e2 100644
--- a/component/block_cache/block_cache.go
+++ b/component/block_cache/block_cache.go
@@ -1190,6 +1190,7 @@ func (bc *BlockCache) lineupDownload(handle *handlemap.Handle, block *Block, pre
 		failCnt:  0,
 		upload:   false,
 		ETag:     Etag,
+		fileSize: handle.Size,
 	}
 
 	// Remove this block from free block list and add to in-process list
@@ -1258,12 +1259,12 @@ func (bc *BlockCache) download(item *workItem) {
 				}
 
 				if numberOfBytes != int(bc.blockSize) &&
-					item.block.offset+uint64(numberOfBytes) != uint64(item.handle.Size) {
+					item.block.offset+uint64(numberOfBytes) != uint64(item.fileSize) {
 					log.Err(
 						"BlockCache::download : Local data retrieved from disk size mismatch, Expected %v, OnDisk %v, fileSize %v",
-						bc.getBlockSize(uint64(item.handle.Size), item.block),
+						bc.getBlockSize(uint64(item.fileSize), item.block),
 						numberOfBytes,
-						item.handle.Size,
+						item.fileSize,
 					)
 					successfulRead = false
 					f.Close()
@@ -1305,8 +1306,7 @@ func (bc *BlockCache) download(item *workItem) {
 	if item.failCnt > MAX_FAIL_CNT {
 		// If we failed to read the data 3 times then just give up
 		log.Err(
-			"BlockCache::download : 3 attempts to download a block have failed %v=>%s (index %v, offset %v)",
-			item.handle.ID,
+			"BlockCache::download : 3 attempts to download a block have failed %s (index %v, offset %v)",
 			item.handle.Path,
 			item.block.id,
 			item.block.offset,
@@ -1319,8 +1319,7 @@ func (bc *BlockCache) download(item *workItem) {
 	if err != nil && err != io.EOF {
 		// Fail to read the data so just reschedule this request
 		log.Err(
-			"BlockCache::download : Failed to read %v=>%s from offset %v [%s]",
-			item.handle.ID,
+			"BlockCache::download : Failed to read %s from offset %v [%s]",
 			item.handle.Path,
 			item.block.id,
 			err.Error(),
@@ -1331,8 +1330,7 @@ func (bc *BlockCache) download(item *workItem) {
 	} else if n == 0 {
 		// No data read so just reschedule this request
 		log.Err(
-			"BlockCache::download : Failed to read %v=>%s from offset %v [0 bytes read]",
-			item.handle.ID,
+			"BlockCache::download : Failed to read %s from offset %v [0 bytes read]",
 			item.handle.Path,
 			item.block.id,
 		)
@@ -1345,8 +1343,7 @@ func (bc *BlockCache) download(item *workItem) {
 	if etag != "" {
 		if item.ETag != "" && item.ETag != etag {
 			log.Err(
-				"BlockCache::download : Blob has changed for %v=>%s (index %v, offset %v)",
-				item.handle.ID,
+				"BlockCache::download : Blob has changed for %s (index %v, offset %v)",
 				item.handle.Path,
 				item.block.id,
 				item.block.offset,
@@ -1600,7 +1597,8 @@ func (bc *BlockCache) getOrCreateBlock(handle *handlemap.Handle, offset uint64)
 		block = node.(*Block)
 
 		// If the block was staged earlier then we are overwriting it here so move it back to cooking queue
-		if block.flags.IsSet(BlockFlagSynced) {
+		// The upload worker sets Synced before it signals completion, so wait for it if it is still uploading
+		if block.flags.IsSet(BlockFlagSynced) && !block.flags.IsSet(BlockFlagUploading) {
 			log.Debug(
 				"BlockCache::getOrCreateBlock : Overwriting back to staged block %v for %v=>%s",
 				block.id,
@@ -1777,6 +1775,7 @@ func (bc *BlockCache) lineupUpload(
 		failCnt:  0,
 		upload:   true,
 		blockId:  id,
+		fileSize: handle.Size,
 	}
 
 	block.Uploading()
@@ -1854,7 +1853,7 @@ func (bc *BlockCache) upload(item *workItem) {
 	flock := bc.fileLocks.Get(fileName)
 	flock.Lock()
 	defer flock.Unlock()
-	blockSize := bc.getBlockSize(uint64(item.handle.Size), item.block)
+	blockSize := bc.getBlockSize(uint64(item.fileSize), item.block)
 	// This block is updated so we need to stage it now
 	err := bc.NextComponent().StageData(internal.StageDataOptions{
 		Name:   item.handle.Path,
diff --git a/component/block_cache/block_cache_test.go b/component/block_cache/block_cache_test.go
index 5ccbbb1b1..9fa5555a5 100644
--- a/component/block_cache/block_cache_test.go
+++ b/component/block_cache/block_cache_test.go
@@ -1445,7 +1445,8 @@ func (suite *blockCacheTestSuite) TestZZZZLazyWrite() {
 	suite.assert.True(handle.Dirty())
 
 	_ = tobj.blockCache.ReleaseFile(internal.ReleaseFileOptions{Handle: handle})
-	time.Sleep(1 * time.Second)
+	// wait for the async close before turning lazy write back off
+	tobj.blockCache.fileCloseOpt.Wait()
 	tobj.blockCache.lazyWrite = false
 
 	// As lazy write is enabled flush shall not upload the file
@@ -3320,6 +3321,121 @@ func (suite *blockCacheTestSuite) TestReadCommittedLastBlocksOverwrite() {
 	suite.assert.Equal(h.Size, int64((15*_1MB)+(_1MB/2)))
 }
 
+// An overwrite must wait for an in-flight upload of the same block. The upload worker marks
+// the block Synced before it clears the dirty bit and signals completion, so a writer that
+// trusts Synced alone can have its dirty bit wiped and its data never uploaded.
+func (suite *blockCacheTestSuite) TestOverwriteWaitsForInFlightUpload() {
+	cfg := "block_cache:\n  block-size-mb: 1\n  mem-size-mb: 20\n  prefetch: 12\n  parallelism: 10"
+	tobj, err := setupPipeline(cfg)
+	defer tobj.cleanupPipeline()
+	suite.assert.NoError(err)
+
+	path := getTestFileName(suite.T().Name())
+	h, err := tobj.blockCache.CreateFile(internal.CreateFileOptions{Name: path, Mode: 0777})
+	suite.assert.NoError(err)
+
+	_, err = tobj.blockCache.WriteFile(
+		&internal.WriteFileOptions{Handle: h, Offset: 0, Data: dataBuff[:10]},
+	)
+	suite.assert.NoError(err)
+	node, found := h.GetValue("0")
+	suite.assert.True(found)
+	block := node.(*Block)
+
+	// Put the block in the state an upload worker leaves it in just before it finishes:
+	// queued for upload and already marked Synced, but still dirty and not yet signalled.
+	h.Lock()
+	block.Uploading()
+	block.flags.Set(BlockFlagUploading)
+	block.flags.Set(BlockFlagSynced)
+	tobj.blockCache.addToCooked(h, block)
+	h.Unlock()
+
+	done := make(chan error, 1)
+	go func() {
+		_, werr := tobj.blockCache.WriteFile(
+			&internal.WriteFileOptions{Handle: h, Offset: 0, Data: dataBuff[10:30]},
+		)
+		done <- werr
+	}()
+
+	select {
+	case <-done:
+		suite.assert.Fail("overwrite did not wait for the in-flight upload to finish")
+		return
+	case <-time.After(200 * time.Millisecond):
+	}
+
+	// Let the simulated upload worker finish
+	block.NoMoreDirty()
+	block.Ready(BlockStatusUploaded)
+
+	select {
+	case err = <-done:
+		suite.assert.NoError(err)
+	case <-time.After(5 * time.Second):
+		suite.assert.Fail("overwrite never completed")
+	}
+	suite.assert.True(block.IsDirty(), "overwrite lost its dirty bit")
+
+	err = tobj.blockCache.ReleaseFile(internal.ReleaseFileOptions{Handle: h})
+	suite.assert.NoError(err)
+
+	data, err := os.ReadFile(filepath.Join(tobj.fake_storage_path, path))
+	suite.assert.NoError(err)
+	suite.assert.Equal(dataBuff[10:30], data)
+}
+
+// An upload must stage the block at the size recorded in the block list when it was lined up.
+// If it reads the live handle size instead, a concurrent write that extends the file makes it
+// stage a full block while the block list still records the short size, and the commit then
+// pads it with a filler block, corrupting the file.
+func (suite *blockCacheTestSuite) TestUploadRacingWriteThatExtendsFile() {
+	cfg := "block_cache:\n  block-size-mb: 1\n  mem-size-mb: 20\n  prefetch: 12\n  parallelism: 10"
+	tobj, err := setupPipeline(cfg)
+	defer tobj.cleanupPipeline()
+	suite.assert.NoError(err)
+
+	for i := range 20 {
+		path := fmt.Sprintf("%s_%d", getTestFileName(suite.T().Name()), i)
+		h, err := tobj.blockCache.CreateFile(internal.CreateFileOptions{Name: path, Mode: 0777})
+		suite.assert.NoError(err)
+
+		_, err = tobj.blockCache.WriteFile(
+			&internal.WriteFileOptions{Handle: h, Offset: 0, Data: dataBuff[:10]},
+		)
+		suite.assert.NoError(err)
+
+		h.Lock()
+		err = tobj.blockCache.stageBlocks(h, 1)
+		lst, _ := h.GetValue("blockList")
+		staged := *lst.(map[int64]*blockInfo)[0]
+		h.Unlock()
+		suite.assert.NoError(err)
+
+		// extend the file while block 0 may still be uploading
+		_, err = tobj.blockCache.WriteFile(
+			&internal.WriteFileOptions{Handle: h, Offset: int64(2 * _1MB), Data: dataBuff[:10]},
+		)
+		suite.assert.NoError(err)
+
+		// wait for the upload of block 0, then compare what was staged with what was recorded
+		h.Lock()
+		tobj.blockCache.waitAndFreeUploadedBlocks(h, 1)
+		h.Unlock()
+		stagedPath := filepath.Join(tobj.fake_storage_path, path) + "_" +
+			strings.ReplaceAll(staged.id, "/", "_")
+		fi, err := os.Stat(stagedPath)
+		suite.assert.NoError(err)
+		if err == nil {
+			suite.assert.Equal(int64(staged.size), fi.Size(), "iteration %d", i)
+		}
+
+		err = tobj.blockCache.ReleaseFile(internal.ReleaseFileOptions{Handle: h})
+		suite.assert.NoError(err)
+	}
+}
+
 // In order for 'go test' to run this suite, we need to create
 // a normal test function and pass our suite to suite.Run
 func TestBlockCacheTestSuite(t *testing.T) {
diff --git a/component/block_cache/threadpool.go b/component/block_cache/threadpool.go
index de0157d9d..e4b397f49 100644
--- a/component/block_cache/threadpool.go
+++ b/component/block_cache/threadpool.go
@@ -62,6 +62,7 @@ type workItem struct {
 	upload   bool              // Flag marking this is a upload request or not
 	blockId  string            // BlockId of the block
 	ETag     string            // Etag of the file before scheduling.
+	fileSize int64             // Size of the file when this item was scheduled
 }
 
 // Reason for storing Etag in workitem struct:
diff --git a/component/block_cache/threadpool_test.go b/component/block_cache/threadpool_test.go
index f42ad427e..a42c6b431 100644
--- a/component/block_cache/threadpool_test.go
+++ b/component/block_cache/threadpool_test.go
@@ -126,8 +126,9 @@ func (suite *threadPoolTestSuite) TestPrioritySchedule() {
 		tp.Schedule(i < 20, &workItem{failCnt: 5})
 	}
 
-	time.Sleep(100 * time.Millisecond)
-	suite.assert.Equal(int32(100), callbackCnt)
+	suite.assert.Eventually(func() bool {
+		return atomic.LoadInt32(&callbackCnt) == 100
+	}, 5*time.Second, 10*time.Millisecond)
 	tp.Stop()
 }
 
@@ -158,9 +159,10 @@ func (suite *threadPoolTestSuite) TestPriorityScheduleWithWriter() {
 		tp.Schedule(i < 20, &workItem{failCnt: 5, upload: true, blockId: "test"})
 	}
 
-	time.Sleep(100 * time.Millisecond)
-	suite.assert.Equal(int32(100), callbackWCnt)
-	suite.assert.Equal(int32(0), callbackRCnt)
+	suite.assert.Eventually(func() bool {
+		return atomic.LoadInt32(&callbackWCnt) == 100
+	}, 5*time.Second, 10*time.Millisecond)
+	suite.assert.Equal(int32(0), atomic.LoadInt32(&callbackRCnt))
 	tp.Stop()
 }
 
diff --git a/component/file_cache/async.go b/component/file_cache/async.go
index 688e2af67..847593484 100644
--- a/component/file_cache/async.go
+++ b/component/file_cache/async.go
@@ -122,6 +122,7 @@ func (fc *FileCache) configureScheduler() error {
 		// add cron callback
 		entryId, err := fc.cronScheduler.AddFunc(window.cronExpr, func() {
 			// Is this a transition from inactive?
+			fc.scheduleLock.Lock()
 			windowCount := fc.activeWindows.Add(1)
 			if windowCount == 1 {
 				// transition to active - open the window
@@ -131,6 +132,7 @@ func (fc *FileCache) configureScheduler() error {
 					window.name,
 				)
 			}
+			fc.scheduleLock.Unlock()
 			log.Info(
 				"FileCache::SchedulerCronFunc : %s (%s) started (numActive=%d)",
 				window.name,
@@ -153,6 +155,7 @@ func (fc *FileCache) configureScheduler() error {
 					return
 				case <-ctx.Done():
 					// Window has completed, update active window count
+					fc.scheduleLock.Lock()
 					windowCount = fc.activeWindows.Add(-1)
 					log.Info(
 						"FileCache::SchedulerCronFunc : %s (%s) ended (numActive=%d)",
@@ -168,6 +171,7 @@ func (fc *FileCache) configureScheduler() error {
 							window.name,
 						)
 					}
+					fc.scheduleLock.Unlock()
 					return
 				}
 			}
@@ -212,6 +216,13 @@ func (fc *FileCache) startScheduler() {
 	fc.cronScheduler.Start()
 }
 
+// uploadWindow returns a channel that is closed while scheduled uploads are allowed
+func (fc *FileCache) uploadWindow() <-chan struct{} {
+	fc.scheduleLock.RLock()
+	defer fc.scheduleLock.RUnlock()
+	return fc.startScheduledUploads
+}
+
 // flock must be locked
 func (fc *FileCache) addPendingOp(name string, value pendingFlags) {
 	log.Trace("FileCache::addPendingOp : %s", name)
@@ -237,7 +248,7 @@ func (fc *FileCache) servicePendingOps() {
 		case <-fc.componentStopping:
 			log.Crit("FileCache::servicePendingOps : Stopping")
 			return
-		case <-fc.startScheduledUploads:
+		case <-fc.uploadWindow():
 			if retryDelay > 0 {
 				select {
 				case <-time.After(retryDelay):
@@ -293,7 +304,7 @@ func (fc *FileCache) runPendingOpCycle() (int, error) {
 		select {
 		case <-fc.componentStopping:
 			return false
-		case <-fc.startScheduledUploads:
+		case <-fc.uploadWindow():
 			name := key.(string)
 			numFilesProcessed++
 			if !fc.updateObject(name, value.(pendingFlags)) {
diff --git a/component/file_cache/file_cache.go b/component/file_cache/file_cache.go
index e94b1c3ae..60753cf16 100644
--- a/component/file_cache/file_cache.go
+++ b/component/file_cache/file_cache.go
@@ -85,9 +85,11 @@ type FileCache struct {
 	fileCloseOpt sync.WaitGroup
 
 	componentStopping     chan struct{}
+	pendingOpsWorker      sync.WaitGroup
 	schedule              WeeklySchedule
 	activeWindows         atomic.Int32
-	startScheduledUploads chan struct{}
+	scheduleLock          sync.RWMutex  // guards startScheduledUploads and window open/close transitions
+	startScheduledUploads chan struct{} // closed while uploads are allowed
 	cronScheduler         *cron.Cron
 }
 
@@ -194,7 +196,7 @@ func (fc *FileCache) Start(ctx context.Context) error {
 	}
 	if fc.syncsPendingOps() {
 		fc.pendingOpAdded = make(chan struct{}, 1)
-		go fc.servicePendingOps()
+		fc.pendingOpsWorker.Go(fc.servicePendingOps)
 	}
 
 	return nil
@@ -206,6 +208,7 @@ func (fc *FileCache) Stop() error {
 
 	// stop async uploads
 	close(fc.componentStopping)
+	fc.pendingOpsWorker.Wait()
 
 	// Stop the cron scheduler and wait for running jobs to complete
 	if fc.cronScheduler != nil {
@@ -1583,12 +1586,13 @@ func (fc *FileCache) openFileInternal(handle *handlemap.Handle, flock *common.Lo
 	}
 
 	handle.UnixFD = uint64(f.Fd())
+	handle.SetFileObject(f)
+	// libfuse reads the file object directly once the handle is marked Cached, so set it first
 	if !fc.offloadIO {
 		handle.Flags.Set(handlemap.HandleFlagCached)
 	}
 
 	log.Info("FileCache::openFileInternal : file=%s, fd=%d", handle.Path, f.Fd())
-	handle.SetFileObject(f)
 
 	//set boolean in isDownloadNeeded value to signal that the file has been downloaded
 	handle.RemoveValue("openFileOptions")
@@ -1905,8 +1909,9 @@ func (fc *FileCache) ReadInBuffer(options *internal.ReadInBufferOptions) (int, e
 	// Update cache policy every 1K operations (includes both read and write) instead
 	options.Handle.Lock()
 	options.Handle.OptCnt++
+	optCnt := options.Handle.OptCnt
 	options.Handle.Unlock()
-	if (options.Handle.OptCnt % defaultCacheUpdateCount) == 0 {
+	if (optCnt % defaultCacheUpdateCount) == 0 {
 		_ = fc.FileUsed(options.Handle.Path)
 	}
 
@@ -1954,8 +1959,9 @@ func (fc *FileCache) WriteFile(options *internal.WriteFileOptions) (int, error)
 	// Update cache policy every 1K operations (includes both read and write) instead
 	options.Handle.Lock()
 	options.Handle.OptCnt++
+	optCnt := options.Handle.OptCnt
 	options.Handle.Unlock()
-	if (options.Handle.OptCnt % defaultCacheUpdateCount) == 0 {
+	if (optCnt % defaultCacheUpdateCount) == 0 {
 		_ = fc.FileUsed(options.Handle.Path)
 	}
 
@@ -2088,7 +2094,7 @@ func (fc *FileCache) flushFileCloud(options internal.FlushFileOptions) error {
 
 	// decide whether to schedule the upload instead
 	select {
-	case <-fc.startScheduledUploads:
+	case <-fc.uploadWindow():
 		// upload now
 	default:
 		// schedule is inactive - push to pendingOps
diff --git a/component/file_cache/file_cache_test.go b/component/file_cache/file_cache_test.go
index 6977502ba..a1b98c3b5 100644
--- a/component/file_cache/file_cache_test.go
+++ b/component/file_cache/file_cache_test.go
@@ -40,6 +40,8 @@ import (
 	"runtime"
 	"strconv"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"syscall"
 	"testing"
 	"time"
@@ -248,6 +250,7 @@ func (suite *fileCacheTestSuite) TestConfig() {
 }
 
 func (suite *fileCacheTestSuite) TestNegativeCacheSize() {
+	defer suite.cleanupTest()
 	var cacheSize float64 = -100
 
 	configStr := fmt.Sprintf(
@@ -270,6 +273,7 @@ func (suite *fileCacheTestSuite) TestNegativeCacheSize() {
 
 func (suite *fileCacheTestSuite) TestDefaultCacheSize() {
 	defer suite.cleanupTest()
+	suite.cleanupTest() // teardown the default file cache generated
 	// Setup
 	config := fmt.Sprintf("file_cache:\n  path: %s\n", suite.cache_path)
 	suite.setupTestHelper(
@@ -1540,6 +1544,7 @@ func (suite *fileCacheTestSuite) TestCreateFileInDir() {
 
 func (suite *fileCacheTestSuite) TestCreateFileCreateEmptyFile() {
 	defer suite.cleanupTest()
+	suite.cleanupTest() // teardown the default file cache generated
 	// Configure to create empty files so we create the file in cloud storage
 	createEmptyFile := true
 	config := fmt.Sprintf(
@@ -1566,6 +1571,7 @@ func (suite *fileCacheTestSuite) TestCreateFileCreateEmptyFile() {
 
 func (suite *fileCacheTestSuite) TestCreateFileInDirCreateEmptyFile() {
 	defer suite.cleanupTest()
+	suite.cleanupTest() // teardown the default file cache generated
 	// Configure to create empty files so we create the file in cloud storage
 	createEmptyFile := true
 	config := fmt.Sprintf(
@@ -1598,6 +1604,7 @@ func (suite *fileCacheTestSuite) TestCreateFileInDirCreateEmptyFile() {
 
 func (suite *fileCacheTestSuite) TestChmodNonexistentCreateEmptyFile() {
 	defer suite.cleanupTest()
+	suite.cleanupTest() // teardown the default file cache generated
 	// Set flag high to test bugfix
 	createEmptyFile := true
 	config := fmt.Sprintf(
@@ -2019,9 +2026,10 @@ func (suite *fileCacheTestSuite) TestOfflineToConnected() {
 	suite.fileCache = newTestFileCache(suite.mock)
 	suite.useMock = true
 
-	connected := false
+	var connected atomic.Bool
+	connected.Store(false)
 	suite.mock.EXPECT().CloudConnected().AnyTimes().DoAndReturn(func() bool {
-		return connected
+		return connected.Load()
 	})
 
 	err = suite.fileCache.Start(context.Background())
@@ -2051,7 +2059,7 @@ func (suite *fileCacheTestSuite) TestOfflineToConnected() {
 	suite.assert.NotNil(handle)
 
 	// Simulate connection restored
-	connected = true
+	connected.Store(true)
 
 	// Write to the file (now connected, but write is purely local)
 	newData := []byte("written after reconnect")
@@ -2089,9 +2097,10 @@ func (suite *fileCacheTestSuite) TestConnectedToOffline() {
 	suite.fileCache = newTestFileCache(suite.mock)
 	suite.useMock = true
 
-	connected := true
+	var connected atomic.Bool
+	connected.Store(true)
 	suite.mock.EXPECT().CloudConnected().AnyTimes().DoAndReturn(func() bool {
-		return connected
+		return connected.Load()
 	})
 
 	err = suite.fileCache.Start(context.Background())
@@ -2147,7 +2156,7 @@ func (suite *fileCacheTestSuite) TestConnectedToOffline() {
 	suite.assert.FileExists(openLocalPath)
 
 	// Simulate connection drop
-	connected = false
+	connected.Store(false)
 
 	// Access the lazy-open file: should fail (data unavailable offline)
 	buf := make([]byte, 10)
@@ -2365,6 +2374,86 @@ func (suite *fileCacheTestSuite) TestReadInBuffer() {
 	suite.assert.Equal(len(data), length)
 }
 
+func (suite *fileCacheTestSuite) TestReadInBufferConcurrentSameHandle() {
+	defer suite.cleanupTest()
+	file := "file_concurrent_read"
+	data := []byte("test data")
+	err := os.WriteFile(filepath.Join(suite.fake_storage_path, file), data, 0777)
+	suite.assert.NoError(err)
+	handle, err := suite.fileCache.OpenFile(
+		internal.OpenFileOptions{Name: file, Flags: os.O_RDONLY, Mode: 0777},
+	)
+	suite.assert.NoError(err)
+
+	var wg sync.WaitGroup
+	for range 8 {
+		wg.Go(func() {
+			output := make([]byte, len(data))
+			for range 200 {
+				n, err := suite.fileCache.ReadInBuffer(
+					&internal.ReadInBufferOptions{Handle: handle, Offset: 0, Data: output},
+				)
+				suite.assert.NoError(err)
+				suite.assert.Equal(len(data), n)
+			}
+		})
+	}
+	wg.Wait()
+
+	err = suite.fileCache.ReleaseFile(internal.ReleaseFileOptions{Handle: handle})
+	suite.assert.NoError(err)
+}
+
+// The first reads of a lazily opened file race with the open that downloads it. Like libfuse,
+// readers use the file object directly once the handle is marked Cached.
+func (suite *fileCacheTestSuite) TestConcurrentFirstReadsOfLazyOpenedFile() {
+	defer suite.cleanupTest()
+	suite.cleanupTest() // teardown the default file cache generated
+	configContent := fmt.Sprintf(
+		"file_cache:\n  path: %s\n  offload-io: false\n\nloopbackfs:\n  path: %s",
+		suite.cache_path,
+		suite.fake_storage_path,
+	)
+	suite.setupTestHelper(configContent)
+
+	data := []byte("lazily opened data")
+	for i := range 50 {
+		name := fmt.Sprintf("lazy_open_%d.txt", i)
+		err := os.WriteFile(filepath.Join(suite.fake_storage_path, name), data, 0777)
+		suite.assert.NoError(err)
+		handle, err := suite.fileCache.OpenFile(
+			internal.OpenFileOptions{Name: name, Flags: os.O_RDONLY, Mode: 0777},
+		)
+		suite.assert.NoError(err)
+		suite.assert.False(handle.Cached())
+
+		// readers that find the handle marked Cached read its file object directly
+		var wg sync.WaitGroup
+		for range 3 {
+			wg.Go(func() {
+				for !handle.Cached() {
+					runtime.Gosched()
+				}
+				buf := make([]byte, len(data))
+				n, err := handle.FObj.ReadAt(buf, 0)
+				suite.assert.NoError(err)
+				suite.assert.Equal(len(data), n)
+			})
+		}
+		// the first read through file cache completes the lazy open
+		buf := make([]byte, len(data))
+		n, err := suite.fileCache.ReadInBuffer(
+			&internal.ReadInBufferOptions{Handle: handle, Data: buf},
+		)
+		suite.assert.NoError(err)
+		suite.assert.Equal(len(data), n)
+		wg.Wait()
+
+		err = suite.fileCache.ReleaseFile(internal.ReleaseFileOptions{Handle: handle})
+		suite.assert.NoError(err)
+	}
+}
+
 func (suite *fileCacheTestSuite) TestReadInBufferErrorBadFd() {
 	defer suite.cleanupTest()
 	// Setup
@@ -2549,6 +2638,7 @@ func (suite *fileCacheTestSuite) TestFlushFileErrorBadFd() {
 
 func (suite *fileCacheTestSuite) TestCronOffToONUpload() {
 	defer suite.cleanupTest()
+	suite.cleanupTest() // teardown the default file cache generated
 
 	// Schedule 3 seconds in the future to allow setup and file creation time
 	now := time.Now()
@@ -2614,6 +2704,7 @@ loopbackfs:
 
 func (suite *fileCacheTestSuite) TestCronOnToOFFUpload() {
 	defer suite.cleanupTest()
+	suite.cleanupTest() // teardown the default file cache generated
 
 	// Schedule for 3 seconds in the future to allow setup time
 	now := time.Now()
@@ -2688,8 +2779,111 @@ loopbackfs:
 	suite.assert.True(scheduled, "File should be scheduled when scheduler is OFF")
 }
 
+// Back-to-back upload windows open and close while files are being flushed
+func (suite *fileCacheTestSuite) TestScheduleWindowTransitionsDuringFlush() {
+	defer suite.cleanupTest()
+	suite.cleanupTest() // teardown the default file cache generated
+	configContent := fmt.Sprintf(`file_cache:
+  path: %s
+  offload-io: true
+  create-empty-file: false
+  schedule:
+    - name: "EverySecond"
+      cron: "* * * * * *"
+      duration: "1s"
+
+loopbackfs:
+  path: %s`,
+		suite.cache_path,
+		suite.fake_storage_path,
+	)
+	suite.setupTestHelper(configContent)
+
+	deadline := time.Now().Add(3500 * time.Millisecond)
+	var wg sync.WaitGroup
+	for w := range 4 {
+		wg.Go(func() {
+			for i := 0; time.Now().Before(deadline); i++ {
+				name := fmt.Sprintf("window_flush_%d_%d.txt", w, i%10)
+				handle, err := suite.fileCache.OpenFile(
+					internal.OpenFileOptions{
+						Name:  name,
+						Flags: os.O_CREATE | os.O_RDWR,
+						Mode:  0777,
+					},
+				)
+				if !suite.assert.NoError(err) {
+					return
+				}
+				_, err = suite.fileCache.WriteFile(
+					&internal.WriteFileOptions{Handle: handle, Data: []byte("data")},
+				)
+				suite.assert.NoError(err)
+				err = suite.fileCache.ReleaseFile(internal.ReleaseFileOptions{Handle: handle})
+				suite.assert.NoError(err)
+				time.Sleep(5 * time.Millisecond)
+			}
+		})
+	}
+	wg.Wait()
+}
+
+// slowUploadComponent delays uploads so a test can stop the file cache mid-upload
+type slowUploadComponent struct {
+	internal.Component
+	started  chan struct{}
+	finished atomic.Bool
+}
+
+func (c *slowUploadComponent) CopyFromFile(options internal.CopyFromFileOptions) error {
+	close(c.started)
+	time.Sleep(500 * time.Millisecond)
+	err := c.Component.CopyFromFile(options)
+	c.finished.Store(true)
+	return err
+}
+
+// Stop must wait for the pending-op worker, so an upload is not still running
+// while the cache directory is cleaned up and the process shuts down
+func (suite *fileCacheTestSuite) TestStopWaitsForPendingOpUpload() {
+	defer suite.cleanupTest()
+	suite.cleanupTest() // teardown the default file cache generated
+	configContent := fmt.Sprintf(
+		"file_cache:\n  path: %s\n  offload-io: true\n\nloopbackfs:\n  path: %s",
+		suite.cache_path,
+		suite.fake_storage_path,
+	)
+	err := config.ReadConfigFromReader(strings.NewReader(configContent))
+	suite.assert.NoError(err)
+	suite.loopback = newLoopbackFS()
+	slow := &slowUploadComponent{Component: suite.loopback, started: make(chan struct{})}
+	suite.fileCache = newTestFileCache(slow)
+	suite.assert.NoError(suite.loopback.Start(context.Background()))
+	suite.assert.NoError(suite.fileCache.Start(context.Background()))
+
+	name := "pending_upload.txt"
+	err = os.WriteFile(filepath.Join(suite.cache_path, name), []byte("data"), 0777)
+	suite.assert.NoError(err)
+	flock := suite.fileCache.fileLocks.Get(name)
+	flock.Lock()
+	suite.fileCache.addPendingOp(name, pendingFlags{})
+	flock.Unlock()
+
+	select {
+	case <-slow.started:
+	case <-time.After(5 * time.Second):
+		suite.assert.Fail("pending upload never started")
+	}
+	suite.assert.NoError(suite.fileCache.Stop())
+	suite.assert.True(slow.finished.Load(), "Stop returned while an upload was still running")
+
+	// give the deferred cleanup a running file cache to stop
+	suite.setupTestHelper(configContent)
+}
+
 func (suite *fileCacheTestSuite) TestNoScheduleAlwaysOn() {
 	defer suite.cleanupTest()
+	suite.cleanupTest() // teardown the default file cache generated
 
 	configContent := fmt.Sprintf(`file_cache:
   path: %s
@@ -2730,6 +2924,7 @@ loopbackfs:
 
 func (suite *fileCacheTestSuite) TestRenamePendingOp() {
 	defer suite.cleanupTest()
+	suite.cleanupTest() // teardown the default file cache generated
 
 	now := time.Now()
 	second := (now.Second() + 30) % 60
@@ -2797,6 +2992,7 @@ loopbackfs:
 
 func (suite *fileCacheTestSuite) TestDeleteScheduledFile() {
 	defer suite.cleanupTest()
+	suite.cleanupTest() // teardown the default file cache generated
 
 	now := time.Now()
 	second := (now.Second() + 30) % 60
@@ -2986,6 +3182,7 @@ func (suite *fileCacheTestSuite) TestServicePendingOpsProcessesPendingOnline() {
 
 func (suite *fileCacheTestSuite) TestCreateEmptyFileEqualTrue() {
 	defer suite.cleanupTest()
+	suite.cleanupTest() // teardown the default file cache generated
 
 	now := time.Now()
 	second := (now.Second() + 30) % 60
@@ -3036,6 +3233,7 @@ loopbackfs:
 
 func (suite *fileCacheTestSuite) TestReadWriteLocalFile() {
 	defer suite.cleanupTest()
+	suite.cleanupTest() // teardown the default file cache generated
 
 	now := time.Now()
 	second := (now.Second() + 30) % 60
@@ -3113,7 +3311,10 @@ loopbackfs:
 }
 
 func (suite *fileCacheTestSuite) TestInvalidCronExpression() {
-	defer suite.cleanupTest()
+	// no file cache is started by this test, so only the default one needs to be stopped
+	suite.cleanupTest() // teardown the default file cache generated
+	defer os.RemoveAll(suite.cache_path)
+	defer os.RemoveAll(suite.fake_storage_path)
 
 	// Set up a configuration with an invalid cron expression
 	configContent := fmt.Sprintf(`file_cache:
@@ -3140,6 +3341,7 @@ loopbackfs:
 
 func (suite *fileCacheTestSuite) TestOverlappingSchedules() {
 	defer suite.cleanupTest()
+	suite.cleanupTest() // teardown the default file cache generated
 
 	now := time.Now()
 	// Create two schedules that will run in close succession (2 seconds apart)
@@ -4066,6 +4268,7 @@ func (suite *fileCacheTestSuite) TestCachePathSymlink() {
 
 func (suite *fileCacheTestSuite) TestZZOffloadIO() {
 	defer suite.cleanupTest()
+	suite.cleanupTest() // teardown the default file cache generated
 	configuration := fmt.Sprintf(
 		"file_cache:\n  path: %s\n  timeout-sec: %d\n\nloopbackfs:\n  path: %s",
 		suite.cache_path,
@@ -4115,6 +4318,7 @@ func (suite *fileCacheTestSuite) TestZZZZLazyWrite() {
 
 func (suite *fileCacheTestSuite) TestStatFS() {
 	defer suite.cleanupTest()
+	suite.cleanupTest() // teardown the default file cache generated
 	cacheTimeout := 5
 	maxSizeMb := 2
 	config := fmt.Sprintf(
@@ -4152,6 +4356,7 @@ func (suite *fileCacheTestSuite) TestStatFS() {
 
 func (suite *fileCacheTestSuite) TestReadFileWithRefresh() {
 	defer suite.cleanupTest()
+	suite.cleanupTest() // teardown the default file cache generated
 	// Configure to create empty files so we create the file in cloud storage
 	config := fmt.Sprintf(
 		"file_cache:\n  path: %s\n  offload-io: true\n  refresh-sec: 1\n\nloopbackfs:\n  path: %s",
@@ -4215,6 +4420,7 @@ func (suite *fileCacheTestSuite) TestReadFileWithRefresh() {
 
 func (suite *fileCacheTestSuite) TestHardLimitOnSize() {
 	defer suite.cleanupTest()
+	suite.cleanupTest() // teardown the default file cache generated
 	// Configure to create empty files so we create the file in cloud storage
 	config := fmt.Sprintf(
 		"file_cache:\n  path: %s\n  offload-io: true\n  hard-limit: true\n  max-size-mb: 2\n\nloopbackfs:\n  path: %s",
@@ -4326,6 +4532,7 @@ func (suite *fileCacheTestSuite) TestHandleDataChange() {
 // are allowed be to deleted but non empty are not
 func (suite *fileCacheTestSuite) TestDeleteDirectory() {
 	defer suite.cleanupTest()
+	suite.cleanupTest() // teardown the default file cache generated
 
 	config := fmt.Sprintf("file_cache:\n  path: %s\n  timeout-sec: 1000\n\nloopbackfs:\n  path: %s",
 		suite.cache_path, suite.fake_storage_path)
diff --git a/component/file_cache/lru_policy.go b/component/file_cache/lru_policy.go
index 497a8196a..4bd3f67c4 100644
--- a/component/file_cache/lru_policy.go
+++ b/component/file_cache/lru_policy.go
@@ -525,14 +525,15 @@ func (p *lruPolicy) setHead(node *lruNode) {
 func (p *lruPolicy) deleteExpiredNodes() {
 	log.Debug("lruPolicy::deleteExpiredNodes : Starts")
 
+	p.Lock()
 	if p.lastMarker.next == nil {
+		p.Unlock()
 		return
 	}
 
 	delItems := make([]*lruNode, 0, p.maxEviction)
 	count := uint32(0)
 
-	p.Lock()
 	node := p.lastMarker.next
 	p.lastMarker.next = nil
 
diff --git a/component/file_cache/lru_policy_test.go b/component/file_cache/lru_policy_test.go
index 9cdb8f046..13045bf67 100644
--- a/component/file_cache/lru_policy_test.go
+++ b/component/file_cache/lru_policy_test.go
@@ -209,6 +209,7 @@ func (suite *lruPolicyTestSuite) TestCacheValid() {
 
 func (suite *lruPolicyTestSuite) TestCachePurge() {
 	defer suite.cleanupTest()
+	suite.cleanupTest() // teardown the default policy generated
 
 	config := cachePolicyConfig{
 		tmpPath:       cache_path,
@@ -356,6 +357,27 @@ func (suite *lruPolicyTestSuite) verifyPolicy(expectedPolicy, actualPolicy *lruP
 	}
 }
 
+// The timeout monitor expires nodes while file operations keep reordering the LRU list
+func (suite *lruPolicyTestSuite) TestCacheValidDuringExpiry() {
+	defer suite.cleanupTest()
+	pathPrefix := filepath.Join(cache_path, "temp")
+	deadline := time.Now().Add(2500 * time.Millisecond)
+	var wg sync.WaitGroup
+	for w := range 4 {
+		wg.Go(func() {
+			for i := w; time.Now().Before(deadline); i++ {
+				suite.policy.CacheValid(pathPrefix + fmt.Sprint(i%2))
+			}
+		})
+	}
+	// run expiry passes the way the timeout monitor does, but much more often
+	for time.Now().Before(deadline) {
+		suite.policy.updateMarker()
+		suite.policy.deleteExpiredNodes()
+	}
+	wg.Wait()
+}
+
 func (suite *lruPolicyTestSuite) TestCreateSnapshotEmpty() {
 	defer suite.cleanupTest()
 	originalPolicy := suite.policy
diff --git a/component/loopback/loopback_fs.go b/component/loopback/loopback_fs.go
index beb98128a..8957309e1 100644
--- a/component/loopback/loopback_fs.go
+++ b/component/loopback/loopback_fs.go
@@ -269,9 +269,11 @@ func (lfs *LoopbackFS) RenameFile(options internal.RenameFileOptions) error {
 	err := os.Rename(oldPath, newPath)
 	handlemap.GetHandles().Range(func(key, value any) bool {
 		handle := value.(*handlemap.Handle)
+		handle.Lock()
 		if handle.Path == options.Src {
 			handle.Path = options.Dst
 		}
+		handle.Unlock()
 		return true
 	})
 	return err
@@ -298,10 +300,10 @@ func (lfs *LoopbackFS) ReadInBuffer(options *internal.ReadInBufferOptions) (int,
 		options.Handle = handlemap.NewHandle(options.Path)
 		options.Handle.Size = options.Size
 	}
-	log.Trace("LoopbackFS::ReadInBuffer : name=%s", options.Handle.Path)
 	f := options.Handle.GetFileObject()
 
 	if f == nil {
+		log.Trace("LoopbackFS::ReadInBuffer : name=%s", options.Handle.Path)
 		f1, err := os.Open(filepath.Join(lfs.path, options.Handle.Path))
 		if err != nil {
 			return 0, nil
@@ -317,6 +319,7 @@ func (lfs *LoopbackFS) ReadInBuffer(options *internal.ReadInBufferOptions) (int,
 
 	options.Handle.RLock()
 	defer options.Handle.RUnlock()
+	log.Trace("LoopbackFS::ReadInBuffer : name=%s", options.Handle.Path)
 
 	n, err := f.ReadAt(options.Data, options.Offset)
 	if err == io.EOF {
diff --git a/component/loopback/loopback_fs_test.go b/component/loopback/loopback_fs_test.go
index ed293ecb6..ba6c5523f 100644
--- a/component/loopback/loopback_fs_test.go
+++ b/component/loopback/loopback_fs_test.go
@@ -31,6 +31,7 @@ import (
 	"os"
 	"path/filepath"
 	"runtime"
+	"sync"
 	"testing"
 
 	"github.com/Seagate/cloudfuse/common"
@@ -319,6 +320,45 @@ func (suite *LoopbackFSTestSuite) TestRenameWriteFileGetAttr() {
 	assert.Equal(int64(5), info.Size())
 }
 
+// Renaming a file rewrites the path of its open handles while they are being read
+func (suite *LoopbackFSTestSuite) TestRenameFileDuringReadInBuffer() {
+	// Windows does not allow renaming a file while it is open
+	if runtime.GOOS == "windows" {
+		fmt.Println("Skipping test on Windows")
+		return
+	}
+	defer suite.cleanupTest()
+	assert := assert.New(suite.T())
+
+	handle, err := suite.lfs.OpenFile(
+		internal.OpenFileOptions{Name: fileLorem, Flags: os.O_RDONLY, Mode: os.FileMode(0644)},
+	)
+	assert.NoError(err)
+
+	renamed := fileLorem + ".renamed"
+	var wg sync.WaitGroup
+	wg.Go(func() {
+		for i := range 100 {
+			src, dst := fileLorem, renamed
+			if i%2 == 1 {
+				src, dst = dst, src
+			}
+			assert.NoError(suite.lfs.RenameFile(internal.RenameFileOptions{Src: src, Dst: dst}))
+		}
+	})
+	data := make([]byte, 20)
+	for range 100 {
+		_, err := suite.lfs.ReadInBuffer(
+			&internal.ReadInBufferOptions{Handle: handle, Offset: 0, Data: data},
+		)
+		assert.NoError(err)
+	}
+	wg.Wait()
+
+	err = suite.lfs.ReleaseFile(internal.ReleaseFileOptions{Handle: handle})
+	assert.NoError(err)
+}
+
 func (suite *LoopbackFSTestSuite) TestReadInBuffer() {
 	defer suite.cleanupTest()
 	assert := assert.New(suite.T())
diff --git a/component/stream/read.go b/component/stream/read.go
index 06be1cb4a..bc3e8ddfb 100644
--- a/component/stream/read.go
+++ b/component/stream/read.go
@@ -94,7 +94,7 @@ func (r *ReadCache) OpenFile(options internal.OpenFileOptions) (*handlemap.Handl
 	}
 	if !r.StreamOnly {
 		handlemap.CreateCacheObject(int64(r.BufferSize), handle)
-		if r.CachedObjects >= r.CachedObjLimit {
+		if atomic.LoadInt32(&r.CachedObjects) >= r.CachedObjLimit {
 			log.Trace(
 				"Stream::OpenFile : file handle limit exceeded - switch handle to stream only mode %s [%v]",
 				options.Name,
diff --git a/component/stream/read_test.go b/component/stream/read_test.go
index c68afbb2c..23b8bd72c 100644
--- a/component/stream/read_test.go
+++ b/component/stream/read_test.go
@@ -826,8 +826,12 @@ func (suite *streamTestSuite) TestAsyncOpen() {
 			0,
 		)
 		suite.mock.EXPECT().OpenFile(openFileOptions).Return(handle, nil)
+		// match on the handle itself: comparing the whole options would read the other
+		// goroutine's handle while it is being modified
 		suite.mock.EXPECT().
-			ReadInBuffer(readInBufferOptions).
+			ReadInBuffer(gomock.Cond(func(o *internal.ReadInBufferOptions) bool {
+				return o.Handle == handle && o.Offset == readInBufferOptions.Offset
+			})).
 			Return(int(suite.stream.BlockSize), nil)
 		wg.Add(1)
 		go asyncOpenFile(suite, openFileOptions)
@@ -858,8 +862,12 @@ func (suite *streamTestSuite) TestAsyncClose() {
 			0,
 		)
 		suite.mock.EXPECT().OpenFile(openFileOptions).Return(handle, nil)
+		// match on the handle itself: comparing the whole options would read the other
+		// goroutine's handle while it is being modified
 		suite.mock.EXPECT().
-			ReadInBuffer(readInBufferOptions).
+			ReadInBuffer(gomock.Cond(func(o *internal.ReadInBufferOptions) bool {
+				return o.Handle == handle && o.Offset == readInBufferOptions.Offset
+			})).
 			Return(int(suite.stream.BlockSize), nil)
 		wg.Add(1)
 		go asyncOpenFile(suite, openFileOptions)
@@ -868,7 +876,11 @@ func (suite *streamTestSuite) TestAsyncClose() {
 
 	for _, handle := range []*handlemap.Handle{handle_1, handle_2} {
 		releaseFileOptions := internal.ReleaseFileOptions{Handle: handle}
-		suite.mock.EXPECT().ReleaseFile(releaseFileOptions).Return(nil)
+		suite.mock.EXPECT().
+			ReleaseFile(gomock.Cond(func(o internal.ReleaseFileOptions) bool {
+				return o.Handle == handle
+			})).
+			Return(nil)
 		wg.Add(1)
 		go asyncCloseFile(suite, releaseFileOptions)
 	}
diff --git a/component/xload/lister.go b/component/xload/lister.go
index f5e90783d..11b2b8cea 100644
--- a/component/xload/lister.go
+++ b/component/xload/lister.go
@@ -193,7 +193,7 @@ func (rl *remoteLister) Process(item *WorkItem) (int, error) {
 				// TODO:: xload : check how many threads can we spawn
 				go func(name string) {
 					localPath := filepath.Join(rl.path, name)
-					err = rl.mkdir(localPath)
+					err := rl.mkdir(localPath)
 					// TODO:: xload : handle error
 					if err != nil {
 						log.Err(
@@ -204,7 +204,7 @@ func (rl *remoteLister) Process(item *WorkItem) (int, error) {
 					}
 
 					// push the directory to input pool for its listing
-					err := rl.Schedule(&WorkItem{
+					err = rl.Schedule(&WorkItem{
 						CompName: rl.GetName(),
 						Path:     name,
 					})
diff --git a/component/xload/splitter_test.go b/component/xload/splitter_test.go
index 6b7d2c3b3..26042aa7e 100644
--- a/component/xload/splitter_test.go
+++ b/component/xload/splitter_test.go
@@ -261,8 +261,10 @@ func (suite *splitterTestSuite) TestSplitterStartStop() {
 
 	time.Sleep(5 * time.Second)
 
-	// stop comoponents
+	// stop components
 	rl.Stop()
+	ds.Stop()
+	rdm.Stop()
 
 	validateMD5(ts.path, remote_path, suite.assert)
 }
@@ -315,8 +317,10 @@ func (suite *splitterTestSuite) TestSplitterConsistency() {
 
 	time.Sleep(5 * time.Second)
 
-	// stop comoponents
+	// stop components
 	rl.Stop()
+	ds.Stop()
+	rdm.Stop()
 
 	validateMD5(ts.path, remote_path, suite.assert)
 }
diff --git a/component/xload/stats_manager.go b/component/xload/stats_manager.go
index 5f74ed87e..df80e750a 100644
--- a/component/xload/stats_manager.go
+++ b/component/xload/stats_manager.go
@@ -53,6 +53,8 @@ type StatsManager struct {
 	waitGroup       sync.WaitGroup  // wait group to wait for stats manager thread to finish
 	items           chan *StatsItem // channel to hold the stats items
 	done            chan bool       // channel to indicate if the stats manager has completed or not
+	stop            chan struct{}   // closed by Stop to end the stats exporter thread
+	exporterDone    sync.WaitGroup  // wait group to wait for the stats exporter thread to finish
 	pool            *BlockPool      // Object of block pool
 }
 
@@ -108,6 +110,7 @@ func NewStatsManager(count uint32, isExportEnabled bool, pool *BlockPool) (*Stat
 		fileHandle: fh,
 		items:      make(chan *StatsItem, count*2),
 		done:       make(chan bool, 1),
+		stop:       make(chan struct{}),
 		pool:       pool,
 	}, nil
 }
@@ -124,16 +127,15 @@ func (sm *StatsManager) Start() {
 	_ = sm.writeToJSON([]byte("\n]"), false)
 
 	go sm.statsProcessor()
-	go sm.statsExporter()
+	sm.exporterDone.Go(sm.statsExporter)
 }
 
 // TODO:: xload : the stop method runs on unmount. See if the channels can be closed if the job is 100% complete
 func (sm *StatsManager) Stop() {
 	log.Debug("statsManager::stop : stop stats manager")
-	sm.done <- true // close the stats exporter thread
-	close(
-		sm.done,
-	) // TODO::xload : check if closing the done channel here will lead to closing the stats exporter thread
+	// stop the exporter before closing items, as it is the processor's only internal sender
+	close(sm.stop)
+	sm.exporterDone.Wait()
 	close(sm.items)
 	sm.waitGroup.Wait()
 
@@ -206,6 +208,9 @@ func (sm *StatsManager) statsExporter() {
 		case <-sm.done:
 			ticker.Stop()
 			return
+		case <-sm.stop:
+			ticker.Stop()
+			return
 		case <-ticker.C:
 			sm.AddStats(&StatsItem{
 				Component: STATS_MANAGER,
@@ -268,7 +273,11 @@ func (sm *StatsManager) calculateBandwidth() {
 
 	// TODO:: xload : determine more effective way to decide if the listing has completed and the stats exporter can be terminated
 	if sm.totalFiles == filesProcessed && sm.totalFiles != sm.dirs {
-		sm.done <- true
+		// the exporter may already have been told to finish, so do not block
+		select {
+		case sm.done <- true:
+		default:
+		}
 		return
 	}
 }
diff --git a/component/xload/threadpool.go b/component/xload/threadpool.go
index 7c1edf313..fc5b533f5 100644
--- a/component/xload/threadpool.go
+++ b/component/xload/threadpool.go
@@ -50,6 +50,10 @@ type ThreadPool struct {
 
 	// Context to cancel the thread pool
 	ctx context.Context
+
+	// Closed by Stop to end the workers and reject new work items.
+	// The item channels are never closed, so a late Schedule can not panic.
+	stop chan struct{}
 }
 
 // NewThreadPool creates a new thread pool
@@ -63,6 +67,7 @@ func NewThreadPool(count uint32, callback func(*WorkItem) (int, error)) *ThreadP
 		callback:      callback,
 		priorityItems: make(chan *WorkItem, count*2),
 		workItems:     make(chan *WorkItem, count*4),
+		stop:          make(chan struct{}),
 	}
 }
 
@@ -81,9 +86,8 @@ func (threadPool *ThreadPool) Start(ctx context.Context) {
 
 // Stop all the workers threads
 func (threadPool *ThreadPool) Stop() {
-	log.Debug("threadPool::Stop : Closing Channels")
-	close(threadPool.priorityItems)
-	close(threadPool.workItems)
+	log.Debug("threadPool::Stop : Stopping workers")
+	close(threadPool.stop)
 	threadPool.waitGroup.Wait()
 	log.Debug("threadPool::Stop : Threads terminated")
 }
@@ -92,22 +96,29 @@ func (threadPool *ThreadPool) Stop() {
 func (threadPool *ThreadPool) Schedule(item *WorkItem) error {
 	// item.Priority specifies the priority of this task.
 	// true means high priority and false means low priority
+	items := threadPool.workItems
+	if item.Priority {
+		items = threadPool.priorityItems
+	}
+
 	select {
 	case <-threadPool.ctx.Done():
-		log.Err(
-			"ThreadPool::Schedule : Thread pool is closed, cannot schedule workitem %s",
-			item.Path,
-		)
-		return fmt.Errorf("thread pool is closed, cannot schedule workitem %s", item.Path)
+	case <-threadPool.stop:
 	default:
-		if item.Priority {
-			threadPool.priorityItems <- item
-		} else {
-			threadPool.workItems <- item
+		// the workers may exit while we wait for queue space, so keep watching for that
+		select {
+		case items <- item:
+			return nil
+		case <-threadPool.ctx.Done():
+		case <-threadPool.stop:
 		}
 	}
 
-	return nil
+	log.Err(
+		"ThreadPool::Schedule : Thread pool is closed, cannot schedule workitem %s",
+		item.Path,
+	)
+	return fmt.Errorf("thread pool is closed, cannot schedule workitem %s", item.Path)
 }
 
 // Do is the core task to be executed by each worker thread
@@ -120,10 +131,9 @@ func (threadPool *ThreadPool) Do(priority bool) {
 			select {
 			case <-threadPool.ctx.Done(): // listen to cancellation signal
 				return
-			case item, ok := <-threadPool.priorityItems:
-				if !ok {
-					return
-				}
+			case <-threadPool.stop:
+				return
+			case item := <-threadPool.priorityItems:
 				threadPool.process(item)
 			}
 		}
@@ -133,15 +143,11 @@ func (threadPool *ThreadPool) Do(priority bool) {
 			select {
 			case <-threadPool.ctx.Done(): // listen to cancellation signal
 				return
-			case item, ok := <-threadPool.priorityItems:
-				if !ok {
-					return
-				}
+			case <-threadPool.stop:
+				return
+			case item := <-threadPool.priorityItems:
 				threadPool.process(item)
-			case item, ok := <-threadPool.workItems:
-				if !ok {
-					return
-				}
+			case item := <-threadPool.workItems:
 				threadPool.process(item)
 			}
 		}
diff --git a/component/xload/threadpool_test.go b/component/xload/threadpool_test.go
index 668520893..183999003 100644
--- a/component/xload/threadpool_test.go
+++ b/component/xload/threadpool_test.go
@@ -126,8 +126,9 @@ func (suite *threadPoolTestSuite) TestPrioritySchedule() {
 
 	}
 
-	time.Sleep(1 * time.Second)
-	suite.assert.Equal(int32(100), callbackCnt)
+	suite.assert.Eventually(func() bool {
+		return atomic.LoadInt32(&callbackCnt) == 100
+	}, 5*time.Second, 10*time.Millisecond)
 	tp.Stop()
 }
 
diff --git a/component/xload/xload_test.go b/component/xload/xload_test.go
index 9364cebc4..5b10d6a6c 100644
--- a/component/xload/xload_test.go
+++ b/component/xload/xload_test.go
@@ -124,7 +124,6 @@ func (suite *xloadTestSuite) setupTestHelper(configuration string, startComponen
 }
 
 func (suite *xloadTestSuite) cleanupTest(stopComp bool) {
-	config.ResetConfig()
 	if stopComp {
 		err := suite.loopback.Stop()
 		suite.assert.NoError(err)
@@ -132,6 +131,8 @@ func (suite *xloadTestSuite) cleanupTest(stopComp bool) {
 		err = suite.xload.Stop()
 		suite.assert.NoError(err)
 	}
+	// reset only after the components stop, as their workers read config
+	config.ResetConfig()
 
 	// Delete the temp directories created
 	os.RemoveAll(suite.local_path)