Fix potential issue with cached sync run persisting after a sync is started or ended. (#612)

* Fix potential issue with cached sync run persisting after a sync is started or ended.

* Remove unneeded bool. Just check if the cached data exists or not.

Author: ggreer

pkg/dotc1z/c1file.go

@@ -47,9 +47,9 @@ type C1File struct {
 	encoderConcurrency int
 
 	// Cached sync run for listConnectorObjects (avoids N+1 queries)
-	cachedViewSyncRun  *syncRun
-	cachedViewSyncOnce sync.Once
-	cachedViewSyncErr  error
+	cachedViewSyncRun *syncRun
+	cachedViewSyncMu  sync.Mutex
+	cachedViewSyncErr error
 
 	// Slow query tracking
 	slowQueryLogTimes     map[string]time.Time

pkg/dotc1z/c1file_test.go

@@ -544,3 +544,65 @@ func TestC1ZReadOnlyMode(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, fileInfo1.ModTime(), fileInfo2.ModTime())
 }
+
+func TestC1ZCachedViewSyncRunInvalidation(t *testing.T) {
+	ctx := t.Context()
+	testFilePath := filepath.Join(c1zTests.workingDir, "test-cached-view-sync-invalidation.c1z")
+
+	f, err := NewC1ZFile(ctx, testFilePath, WithPragma("journal_mode", "WAL"))
+	require.NoError(t, err)
+
+	// Start first sync and add a resource
+	syncID1, err := f.StartNewSync(ctx, connectorstore.SyncTypeFull, "")
+	require.NoError(t, err)
+	require.NotEmpty(t, syncID1)
+
+	err = f.PutResourceTypes(ctx, v2.ResourceType_builder{Id: testResourceType}.Build())
+	require.NoError(t, err)
+
+	err = f.PutResources(ctx, v2.Resource_builder{
+		Id: v2.ResourceId_builder{
+			ResourceType: testResourceType,
+			Resource:     "resource-1",
+		}.Build(),
+	}.Build())
+	require.NoError(t, err)
+
+	err = f.EndSync(ctx)
+	require.NoError(t, err)
+
+	// Call ListResources to populate the cache with sync1
+	resp1, err := f.ListResources(ctx, v2.ResourcesServiceListResourcesRequest_builder{}.Build())
+	require.NoError(t, err)
+	require.Len(t, resp1.GetList(), 1)
+	require.Equal(t, "resource-1", resp1.GetList()[0].GetId().GetResource())
+
+	// Start a new sync and add a different resource
+	syncID2, err := f.StartNewSync(ctx, connectorstore.SyncTypeFull, "")
+	require.NoError(t, err)
+	require.NotEmpty(t, syncID2)
+	require.NotEqual(t, syncID1, syncID2)
+
+	err = f.PutResources(ctx, v2.Resource_builder{
+		Id: v2.ResourceId_builder{
+			ResourceType: testResourceType,
+			Resource:     "resource-2",
+		}.Build(),
+	}.Build())
+	require.NoError(t, err)
+
+	// End the new sync
+	err = f.EndSync(ctx)
+	require.NoError(t, err)
+
+	// Call ListResources again - it should return resource-2 from the new finished sync (sync2),
+	// but it will return resource-1 from the cached sync (sync1) instead because the cache wasn't invalidated
+	resp2, err := f.ListResources(ctx, v2.ResourcesServiceListResourcesRequest_builder{}.Build())
+	require.NoError(t, err)
+	// This assertion will fail because the cache wasn't invalidated when sync2 finished
+	require.Len(t, resp2.GetList(), 1, "should return resource from new sync")
+	require.Equal(t, "resource-2", resp2.GetList()[0].GetId().GetResource(), "should return resource-2 from the new finished sync (sync2), not resource-1 from cached sync (sync1)")
+
+	err = f.Close()
+	require.NoError(t, err)
+}

pkg/dotc1z/sync_runs.go

@@ -97,27 +97,40 @@ type syncRun struct {
 
 // getCachedViewSyncRun returns the cached sync run for read operations.
 // This avoids N+1 queries when paginating through listConnectorObjects.
-// The result is computed once and cached for the lifetime of the C1File.
+// The cache is invalidated when a sync starts or ends.
 func (c *C1File) getCachedViewSyncRun(ctx context.Context) (*syncRun, error) {
 	ctx, span := tracer.Start(ctx, "C1File.getCachedViewSyncRun")
 	defer span.End()
 
-	c.cachedViewSyncOnce.Do(func() {
-		// First try to get a finished full sync
-		c.cachedViewSyncRun, c.cachedViewSyncErr = c.getFinishedSync(ctx, 0, connectorstore.SyncTypeFull)
-		if c.cachedViewSyncErr != nil {
-			return
-		}
+	c.cachedViewSyncMu.Lock()
+	defer c.cachedViewSyncMu.Unlock()
 
-		// If no finished sync, try to get an unfinished one
-		if c.cachedViewSyncRun == nil {
-			c.cachedViewSyncRun, c.cachedViewSyncErr = c.getLatestUnfinishedSync(ctx, connectorstore.SyncTypeAny)
-		}
-	})
+	if c.cachedViewSyncRun != nil || c.cachedViewSyncErr != nil {
+		return c.cachedViewSyncRun, c.cachedViewSyncErr
+	}
+
+	// First try to get a finished full sync
+	c.cachedViewSyncRun, c.cachedViewSyncErr = c.getFinishedSync(ctx, 0, connectorstore.SyncTypeFull)
+	if c.cachedViewSyncErr != nil {
+		return c.cachedViewSyncRun, c.cachedViewSyncErr
+	}
+
+	// If no finished sync, try to get an unfinished one
+	if c.cachedViewSyncRun == nil {
+		c.cachedViewSyncRun, c.cachedViewSyncErr = c.getLatestUnfinishedSync(ctx, connectorstore.SyncTypeAny)
+	}
 
 	return c.cachedViewSyncRun, c.cachedViewSyncErr
 }
 
+// invalidateCachedViewSyncRun clears the cached sync run so it will be recomputed on next access.
+func (c *C1File) invalidateCachedViewSyncRun() {
+	c.cachedViewSyncMu.Lock()
+	defer c.cachedViewSyncMu.Unlock()
+	c.cachedViewSyncRun = nil
+	c.cachedViewSyncErr = nil
+}
+
 func (c *C1File) getLatestUnfinishedSync(ctx context.Context, syncType connectorstore.SyncType) (*syncRun, error) {
 	ctx, span := tracer.Start(ctx, "C1File.getLatestUnfinishedSync")
 	defer span.End()
@@ -539,6 +552,7 @@ func (c *C1File) StartNewSync(ctx context.Context, syncType connectorstore.SyncT
 	}
 
 	c.currentSyncID = syncID
+	c.invalidateCachedViewSyncRun()
 
 	return c.currentSyncID, nil
 }
@@ -597,6 +611,7 @@ func (c *C1File) EndSync(ctx context.Context) error {
 	}
 
 	c.currentSyncID = ""
+	c.invalidateCachedViewSyncRun()
 
 	return nil
 }