Repository: kopia/kopia
Pull request #3666 (2024-02-21 04:56:33)

fix(server): do not make blocking calls inside server status API
also reduce global server lock scope

This API call is frequently invoked by KopiaUI. This should improve #2226 and #2453  by never blocking in the critical API call. 


codecov[bot]: ## [Codecov](https://app.codecov.io/gh/kopia/kopia/pull/3666?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia) Report
Attention: `4 lines` in your changes are missing coverage. Please review.
> Comparison is base [(`cb455c6`)](https://app.codecov.io/gh/kopia/kopia/commit/cb455c699a04e54cb47e25a676d7d65063ce883e?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia) 75.86% compared to head [(`619e75f`)](https://app.codecov.io/gh/kopia/kopia/pull/3666?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia) 76.85%.
> Report is 23 commits behind head on master.

| [Files](https://app.codecov.io/gh/kopia/kopia/pull/3666?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia) | Patch % | Lines |
|---|---|---|
| [internal/server/server.go](https://app.codecov.io/gh/kopia/kopia/pull/3666?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia#diff-aW50ZXJuYWwvc2VydmVyL3NlcnZlci5nbw==) | 80.00% | [2 Missing :warning: ](https://app.codecov.io/gh/kopia/kopia/pull/3666?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia) |
| [repo/format/format\_manager.go](https://app.codecov.io/gh/kopia/kopia/pull/3666?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia#diff-cmVwby9mb3JtYXQvZm9ybWF0X21hbmFnZXIuZ28=) | 66.66% | [1 Missing and 1 partial :warning: ](https://app.codecov.io/gh/kopia/kopia/pull/3666?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia) |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master    #3666      +/-   ##
==========================================
+ Coverage   75.86%   76.85%   +0.98%     
==========================================
  Files         470      471       +1     
  Lines       37301    28594    -8707     
==========================================
- Hits        28299    21976    -6323     
+ Misses       7071     4691    -2380     
+ Partials     1931     1927       -4     
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/kopia/kopia/pull/3666?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia).


Patch:
diff --git a/app/public/server.js b/app/public/server.js
index 15a03617..fb4aea85 100644
--- a/app/public/server.js
+++ b/app/public/server.js
@@ -67,6 +67,8 @@ function newServerForRepo(repoID) {
 
             const statusUpdated = this.raiseStatusUpdatedEvent.bind(this);
 
+	    const pollInterval = 3000; 
+
             function pollOnce() {
                 if (!runningServerAddress || !runningServerCertificate || !runningServerPassword || !runningServerControlPassword) {
                     return;
@@ -78,6 +80,7 @@ function newServerForRepo(repoID) {
                     port: parseInt(new URL(runningServerAddress).port),
                     method: "GET",
                     path: "/api/v1/control/status",
+		    timeout: pollInterval,
                     headers: {
                         'Authorization': 'Basic ' + Buffer.from("server-control" + ':' + runningServerControlPassword).toString('base64')
                      }  
@@ -104,7 +107,7 @@ function newServerForRepo(repoID) {
                 req.end();
             }
 
-            const statusPollInterval = setInterval(pollOnce, 3000);
+            const statusPollInterval = setInterval(pollOnce, pollInterval);
 
             runningServerProcess.on('close', (code, signal) => {
                 this.appendToLog(`child process exited with code ${code} and signal ${signal}`);
diff --git a/internal/server/api_repo.go b/internal/server/api_repo.go
index 5223d592..bcae9a9b 100644
--- a/internal/server/api_repo.go
+++ b/internal/server/api_repo.go
@@ -35,16 +35,11 @@ func handleRepoParameters(ctx context.Context, rc requestContext) (interface{},
 		}, nil
 	}
 
-	scc, err := dr.ContentReader().SupportsContentCompression(ctx)
-	if err != nil {
-		return nil, internalServerError(err)
-	}
-
 	rp := &remoterepoapi.Parameters{
 		HashFunction:               dr.ContentReader().ContentFormat().GetHashFunction(),
 		HMACSecret:                 dr.ContentReader().ContentFormat().GetHmacSecret(),
 		ObjectFormat:               dr.ObjectFormat(),
-		SupportsContentCompression: scc,
+		SupportsContentCompression: dr.ContentReader().SupportsContentCompression(),
 	}
 
 	return rp, nil
@@ -62,15 +57,8 @@ func handleRepoStatus(ctx context.Context, rc requestContext) (interface{}, *api
 	if ok {
 		contentFormat := dr.ContentReader().ContentFormat()
 
-		mp, mperr := contentFormat.GetMutableParameters(ctx)
-		if mperr != nil {
-			return nil, internalServerError(mperr)
-		}
-
-		scc, err := dr.ContentReader().SupportsContentCompression(ctx)
-		if err != nil {
-			return nil, internalServerError(err)
-		}
+		// this gets potentially stale parameters
+		mp := contentFormat.GetCachedMutableParameters()
 
 		return &serverapi.StatusResponse{
 			Connected:                  true,
@@ -84,7 +72,7 @@ func handleRepoStatus(ctx context.Context, rc requestContext) (interface{}, *api
 			Splitter:                   dr.ObjectFormat().Splitter,
 			Storage:                    dr.BlobReader().ConnectionInfo().Type,
 			ClientOptions:              dr.ClientOptions(),
-			SupportsContentCompression: scc,
+			SupportsContentCompression: dr.ContentReader().SupportsContentCompression(),
 		}, nil
 	}
 
diff --git a/internal/server/grpc_session.go b/internal/server/grpc_session.go
index b8d020c3..b0b4a53e 100644
--- a/internal/server/grpc_session.go
+++ b/internal/server/grpc_session.go
@@ -551,10 +551,7 @@ func (s *Server) handleInitialSessionHandshake(srv grpcapi.KopiaRepository_Sessi
 		return repo.WriteSessionOptions{}, errors.Errorf("missing initialization request")
 	}
 
-	scc, err := dr.ContentReader().SupportsContentCompression(srv.Context())
-	if err != nil {
-		return repo.WriteSessionOptions{}, errors.Wrap(err, "supports content compression")
-	}
+	scc := dr.ContentReader().SupportsContentCompression()
 
 	if err := s.send(srv, initializeReq.GetRequestId(), &grpcapi.SessionResponse{
 		Response: &grpcapi.SessionResponse_InitializeSession{
diff --git a/internal/server/server.go b/internal/server/server.go
index 47018cd8..e964487d 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -103,7 +103,9 @@ type Server struct {
 	// +checklocks:serverMutex
 	sched *scheduler.Scheduler
 
-	// +checklocks:serverMutex
+	nextRefreshTimeLock sync.Mutex
+
+	// +checklocks:nextRefreshTimeLock
 	nextRefreshTime time.Time
 
 	grpcServerState
@@ -268,8 +270,8 @@ func (s *Server) generateShortTermAuthCookie(username string, now time.Time) (st
 }
 
 func (s *Server) captureRequestContext(w http.ResponseWriter, r *http.Request) requestContext {
-	s.serverMutex.Lock()
-	defer s.serverMutex.Unlock()
+	s.serverMutex.RLock()
+	defer s.serverMutex.RUnlock()
 
 	return requestContext{
 		w:   w,
@@ -433,9 +435,9 @@ func (s *Server) handleRequestPossiblyNotConnected(isAuthorized isAuthorizedFunc
 
 func (s *Server) refreshAsync() {
 	// prevent refresh from being runnable.
-	s.serverMutex.Lock()
+	s.nextRefreshTimeLock.Lock()
 	s.nextRefreshTime = clock.Now().Add(s.options.RefreshInterval)
-	s.serverMutex.Unlock()
+	s.nextRefreshTimeLock.Unlock()
 
 	go s.Refresh()
 }
@@ -458,7 +460,9 @@ func (s *Server) refreshLocked(ctx context.Context) error {
 		return nil
 	}
 
+	s.nextRefreshTimeLock.Lock()
 	s.nextRefreshTime = clock.Now().Add(s.options.RefreshInterval)
+	s.nextRefreshTimeLock.Unlock()
 
 	if err := s.rep.Refresh(ctx); err != nil {
 		return errors.Wrap(err, "unable to refresh repository")
@@ -1021,11 +1025,15 @@ func (s *Server) getSchedulerItems(ctx context.Context, now time.Time) []schedul
 
 	var result []scheduler.Item
 
+	s.nextRefreshTimeLock.Lock()
+	nrt := s.nextRefreshTime
+	s.nextRefreshTimeLock.Unlock()
+
 	// add a scheduled item to refresh all sources and policies
 	result = append(result, scheduler.Item{
 		Description: "refresh",
 		Trigger:     s.refreshAsync,
-		NextTime:    s.nextRefreshTime,
+		NextTime:    nrt,
 	})
 
 	if s.maint != nil {
diff --git a/repo/api_server_repository.go b/repo/api_server_repository.go
index 0d9b6e2e..46cd379f 100644
--- a/repo/api_server_repository.go
+++ b/repo/api_server_repository.go
@@ -157,8 +157,8 @@ func (r *apiServerRepository) Flush(ctx context.Context) error {
 	return nil
 }
 
-func (r *apiServerRepository) SupportsContentCompression(ctx context.Context) (bool, error) {
-	return r.serverSupportsContentCompression, nil
+func (r *apiServerRepository) SupportsContentCompression() bool {
+	return r.serverSupportsContentCompression
 }
 
 func (r *apiServerRepository) NewWriter(ctx context.Context, opt WriteSessionOptions) (context.Context, RepositoryWriter, error) {
diff --git a/repo/content/content_manager.go b/repo/content/content_manager.go
index bbf0477e..0222a7fe 100644
--- a/repo/content/content_manager.go
+++ b/repo/content/content_manager.go
@@ -768,13 +768,10 @@ func (bm *WriteManager) getOrCreatePendingPackInfoLocked(ctx context.Context, pr
 }
 
 // SupportsContentCompression returns true if content manager supports content-compression.
-func (bm *WriteManager) SupportsContentCompression(ctx context.Context) (bool, error) {
-	mp, mperr := bm.format.GetMutableParameters(ctx)
-	if mperr != nil {
-		return false, errors.Wrap(mperr, "mutable parameters")
-	}
+func (bm *WriteManager) SupportsContentCompression() bool {
+	mp := bm.format.GetCachedMutableParameters()
 
-	return mp.IndexVersion >= index.Version2, nil
+	return mp.IndexVersion >= index.Version2
 }
 
 // WriteContent saves a given content of data to a pack group with a provided name and returns a contentID
diff --git a/repo/content/content_reader.go b/repo/content/content_reader.go
index 0b1cfeae..91af67b7 100644
--- a/repo/content/content_reader.go
+++ b/repo/content/content_reader.go
@@ -9,7 +9,10 @@
 
 // Reader defines content read API.
 type Reader interface {
-	SupportsContentCompression(ctx context.Context) (bool, error)
+	// returns true if the repository supports content compression.
+	// this may be slightly stale if the repository recently
+	// got upgraded, in which case it will return false which is safe.
+	SupportsContentCompression() bool
 	ContentFormat() format.Provider
 	GetContent(ctx context.Context, id ID) ([]byte, error)
 	ContentInfo(ctx context.Context, id ID) (Info, error)
diff --git a/repo/format/content_format.go b/repo/format/content_format.go
index 0dfbe06c..11ee9223 100644
--- a/repo/format/content_format.go
+++ b/repo/format/content_format.go
@@ -50,6 +50,11 @@ func (f *ContentFormat) GetMutableParameters(ctx context.Context) (MutableParame
 	return f.MutableParameters, nil
 }
 
+// GetCachedMutableParameters implements FormattingOptionsProvider.
+func (f *ContentFormat) GetCachedMutableParameters() MutableParameters {
+	return f.MutableParameters
+}
+
 // SupportsPasswordChange implements FormattingOptionsProvider.
 func (f *ContentFormat) SupportsPasswordChange() bool {
 	return f.EnablePasswordChange
diff --git a/repo/format/format_manager.go b/repo/format/format_manager.go
index 9c7d75e3..0d46b8b3 100644
--- a/repo/format/format_manager.go
+++ b/repo/format/format_manager.go
@@ -266,6 +266,18 @@ func (m *Manager) GetMutableParameters(ctx context.Context) (MutableParameters,
 	return f.GetMutableParameters(ctx)
 }
 
+// GetCachedMutableParameters gets mutable paramers of the repository without blocking.
+func (m *Manager) GetCachedMutableParameters() MutableParameters {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+
+	if m.current == nil {
+		return MutableParameters{}
+	}
+
+	return m.current.GetCachedMutableParameters()
+}
+
 // UpgradeLockIntent returns the current lock intent.
 func (m *Manager) UpgradeLockIntent(ctx context.Context) (*UpgradeLockIntent, error) {
 	if err := m.maybeRefreshNotLocked(ctx); err != nil {
diff --git a/repo/format/format_provider.go b/repo/format/format_provider.go
index 3042b137..bb1db2da 100644
--- a/repo/format/format_provider.go
+++ b/repo/format/format_provider.go
@@ -59,6 +59,7 @@ type Provider interface {
 	// this is typically cached, but sometimes refreshes MutableParameters from
 	// the repository so the results should not be cached.
 	GetMutableParameters(ctx context.Context) (MutableParameters, error)
+	GetCachedMutableParameters() MutableParameters
 	SupportsPasswordChange() bool
 	GetMasterKey() []byte
 
diff --git a/repo/grpc_repository_client.go b/repo/grpc_repository_client.go
index 961228f4..5fee712d 100644
--- a/repo/grpc_repository_client.go
+++ b/repo/grpc_repository_client.go
@@ -682,8 +682,8 @@ func (r *grpcInnerSession) GetContent(ctx context.Context, contentID content.ID)
 	return nil, errNoSessionResponse()
 }
 
-func (r *grpcRepositoryClient) SupportsContentCompression(ctx context.Context) (bool, error) {
-	return r.serverSupportsContentCompression, nil
+func (r *grpcRepositoryClient) SupportsContentCompression() bool {
+	return r.serverSupportsContentCompression
 }
 
 func (r *grpcRepositoryClient) doWriteAsync(ctx context.Context, contentID content.ID, data []byte, prefix content.IDPrefix, comp compression.HeaderID) error {
diff --git a/repo/object/object_manager.go b/repo/object/object_manager.go
index 5ff04526..dfc16a96 100644
--- a/repo/object/object_manager.go
+++ b/repo/object/object_manager.go
@@ -35,7 +35,8 @@ type contentReader interface {
 
 type contentManager interface {
 	contentReader
-	SupportsContentCompression(ctx context.Context) (bool, error)
+
+	SupportsContentCompression() bool
 	WriteContent(ctx context.Context, data gather.Bytes, prefix content.IDPrefix, comp compression.HeaderID) (content.ID, error)
 }
 
diff --git a/repo/object/object_writer.go b/repo/object/object_writer.go
index e4f8b69e..a1f2c874 100644
--- a/repo/object/object_writer.go
+++ b/repo/object/object_writer.go
@@ -188,13 +188,11 @@ func (w *objectWriter) prepareAndWriteContentChunk(chunkID int, data gather.Byte
 	comp := content.NoCompression
 	objectComp := w.compressor
 
-	scc, err := w.om.contentMgr.SupportsContentCompression(w.ctx)
-	if err != nil {
-		return errors.Wrap(err, "supports content compression")
-	}
+	// in super rare cases this may be stale, but if it is it will be false which is always safe.
+	supportsContentCompression := w.om.contentMgr.SupportsContentCompression()
 
 	// do not compress in this layer, instead pass comp to the content manager.
-	if scc && w.compressor != nil {
+	if supportsContentCompression && w.compressor != nil {
 		comp = w.compressor.HeaderID()
 		objectComp = nil
 	}


Test patch:
diff --git a/repo/content/content_manager_test.go b/repo/content/content_manager_test.go
index eeb07723..a68e1a0b 100644
--- a/repo/content/content_manager_test.go
+++ b/repo/content/content_manager_test.go
@@ -1830,7 +1830,7 @@ func (s *contentManagerSuite) TestAutoCompressionOfMetadata(t *testing.T) {
 	info, err := bm.ContentInfo(ctx, contentID)
 	require.NoError(t, err)
 
-	if scc, _ := bm.SupportsContentCompression(ctx); scc {
+	if bm.SupportsContentCompression() {
 		require.Equal(t, compression.HeaderZstdFastest, info.GetCompressionHeaderID())
 	} else {
 		require.Equal(t, NoCompression, info.GetCompressionHeaderID())
diff --git a/repo/object/object_manager_test.go b/repo/object/object_manager_test.go
index 4372f8ff..6785d8ff 100644
--- a/repo/object/object_manager_test.go
+++ b/repo/object/object_manager_test.go
@@ -79,8 +79,8 @@ func (f *fakeContentManager) WriteContent(ctx context.Context, data gather.Bytes
 	return contentID, nil
 }
 
-func (f *fakeContentManager) SupportsContentCompression(ctx context.Context) (bool, error) {
-	return f.supportsContentCompression, nil
+func (f *fakeContentManager) SupportsContentCompression() bool {
+	return f.supportsContentCompression
 }
 
 func (f *fakeContentManager) ContentInfo(ctx context.Context, contentID content.ID) (content.Info, error) {