Repository: kopia/kopia
Pull request #4358 (2025-01-22 23:31:04)

fix(server): ensure server uploads "repodiag" blobs
The kopia server was not uploading any logs to the repository, because _"repodiag"_ blob uploads would always fail.

The cause was the following: When the (log) repodiag blob PUT operation was initiated, the `Context` used for this operation was already canceled.

The context used for blob uploads is passed to `repodiag.NewLogManager` when opening the repository. In the case of the kopia server, the repository is asynchronously opened in `server.Server.InitReposotoryAsync`. The context passed to `repo.Open` is canceled after the "open repository" server task completes. This issue was introduced in #1691

Change:
Use `context.WithoutCancel()` instead of the context passed when the repo "diagnoser" is created.

New tests are included to reproduce this failure and verify the fix.

Ref:
- #1691

Thanks to:
- 🎖️ @redgoat650 for reporting this issue;
- 🎖️ @aaron-kasten for the triaging, testing and reproduction of the issue.

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/kopia/kopia/pull/4358?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 76.30%. Comparing base [(`cb455c6`)](https://app.codecov.io/gh/kopia/kopia/commit/cb455c699a04e54cb47e25a676d7d65063ce883e?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia) to head [(`954dee4`)](https://app.codecov.io/gh/kopia/kopia/commit/954dee494e802e4ad388a1d77578dabf6a6b42ca?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia).
> Report is 400 commits behind head on master.

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


```diff
@@            Coverage Diff             @@
##           master    #4358      +/-   ##
==========================================
+ Coverage   75.86%   76.30%   +0.43%     
==========================================
  Files         470      527      +57     
  Lines       37301    39920    +2619     
==========================================
+ Hits        28299    30460    +2161     
- Misses       7071     7451     +380     
- Partials     1931     2009      +78     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/kopia/kopia/pull/4358?dropdown=coverage&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).

julio-lopez: FYI @jkowalski

Patch:
diff --git a/internal/repodiag/log_manager.go b/internal/repodiag/log_manager.go
index 2ee78485..c05f260c 100644
--- a/internal/repodiag/log_manager.go
+++ b/internal/repodiag/log_manager.go
@@ -78,7 +78,7 @@ func (m *LogManager) Enable() {
 // NewLogManager creates a new LogManager that will emit logs as repository blobs.
 func NewLogManager(ctx context.Context, w *BlobWriter) *LogManager {
 	return &LogManager{
-		ctx:            ctx,
+		ctx:            context.WithoutCancel(ctx),
 		writer:         w,
 		flushThreshold: blobLoggerFlushThreshold,
 		timeFunc:       clock.Now,


Test patch:
diff --git a/internal/blobtesting/map.go b/internal/blobtesting/map.go
index b5f4ccba..db06376b 100644
--- a/internal/blobtesting/map.go
+++ b/internal/blobtesting/map.go
@@ -33,6 +33,10 @@ type mapStorage struct {
 }
 
 func (s *mapStorage) GetCapacity(ctx context.Context) (blob.Capacity, error) {
+	if err := ctx.Err(); err != nil {
+		return blob.Capacity{}, errors.Wrap(err, "get capacity failed")
+	}
+
 	if s.limit < 0 {
 		return blob.Capacity{}, blob.ErrNotAVolume
 	}
@@ -47,6 +51,10 @@ func (s *mapStorage) GetCapacity(ctx context.Context) (blob.Capacity, error) {
 }
 
 func (s *mapStorage) GetBlob(ctx context.Context, id blob.ID, offset, length int64, output blob.OutputBuffer) error {
+	if err := ctx.Err(); err != nil {
+		return errors.Wrap(err, "get blob failed")
+	}
+
 	s.mutex.RLock()
 	defer s.mutex.RUnlock()
 
@@ -82,6 +90,10 @@ func (s *mapStorage) GetBlob(ctx context.Context, id blob.ID, offset, length int
 }
 
 func (s *mapStorage) GetMetadata(ctx context.Context, id blob.ID) (blob.Metadata, error) {
+	if err := ctx.Err(); err != nil {
+		return blob.Metadata{}, errors.Wrap(err, "get metadata failed")
+	}
+
 	s.mutex.RLock()
 	defer s.mutex.RUnlock()
 
@@ -98,6 +110,10 @@ func (s *mapStorage) GetMetadata(ctx context.Context, id blob.ID) (blob.Metadata
 }
 
 func (s *mapStorage) PutBlob(ctx context.Context, id blob.ID, data blob.Bytes, opts blob.PutOptions) error {
+	if err := ctx.Err(); err != nil {
+		return errors.Wrap(err, "pub blob failed")
+	}
+
 	switch {
 	case opts.HasRetentionOptions():
 		return errors.Wrap(blob.ErrUnsupportedPutBlobOption, "blob-retention")
@@ -134,6 +150,10 @@ func (s *mapStorage) PutBlob(ctx context.Context, id blob.ID, data blob.Bytes, o
 }
 
 func (s *mapStorage) DeleteBlob(ctx context.Context, id blob.ID) error {
+	if err := ctx.Err(); err != nil {
+		return errors.Wrap(err, "delete blob failed")
+	}
+
 	s.mutex.Lock()
 	defer s.mutex.Unlock()
 
@@ -145,6 +165,10 @@ func (s *mapStorage) DeleteBlob(ctx context.Context, id blob.ID) error {
 }
 
 func (s *mapStorage) ListBlobs(ctx context.Context, prefix blob.ID, callback func(blob.Metadata) error) error {
+	if err := ctx.Err(); err != nil {
+		return errors.Wrap(err, "list blobs failed")
+	}
+
 	s.mutex.RLock()
 
 	keys := []blob.ID{}
@@ -184,6 +208,10 @@ func (s *mapStorage) ListBlobs(ctx context.Context, prefix blob.ID, callback fun
 }
 
 func (s *mapStorage) TouchBlob(ctx context.Context, blobID blob.ID, threshold time.Duration) (time.Time, error) {
+	if err := ctx.Err(); err != nil {
+		return time.Time{}, errors.Wrap(err, "touch blob failed")
+	}
+
 	s.mutex.Lock()
 	defer s.mutex.Unlock()
 
diff --git a/internal/repodiag/log_manager_test.go b/internal/repodiag/log_manager_test.go
index 47530fbd..4cf79719 100644
--- a/internal/repodiag/log_manager_test.go
+++ b/internal/repodiag/log_manager_test.go
@@ -1,6 +1,7 @@
 package repodiag_test
 
 import (
+	"context"
 	"crypto/rand"
 	"encoding/hex"
 	"strings"
@@ -84,6 +85,30 @@ func TestLogManager_NotEnabled(t *testing.T) {
 	require.Empty(t, d)
 }
 
+func TestLogManager_CancelledContext(t *testing.T) {
+	d := blobtesting.DataMap{}
+	st := blobtesting.NewMapStorage(d, nil, nil)
+	w := repodiag.NewWriter(st, newStaticCrypter(t))
+	ctx := testlogging.Context(t)
+	cctx, cancel := context.WithCancel(ctx)
+	lm := repodiag.NewLogManager(cctx, w)
+
+	// cancel context, logs should still be written
+	cancel()
+
+	lm.Enable()
+	l := lm.NewLogger()
+	l.Info("hello")
+
+	require.Empty(t, d)
+
+	l.Sync()
+	w.Wait(ctx)
+
+	// make sure log messages are written
+	require.Len(t, d, 1)
+}
+
 func TestLogManager_Null(t *testing.T) {
 	var lm *repodiag.LogManager
 
diff --git a/tests/end_to_end_test/server_repo_logs_test.go b/tests/end_to_end_test/server_repo_logs_test.go
new file mode 100644
index 00000000..163b7416
--- /dev/null
+++ b/tests/end_to_end_test/server_repo_logs_test.go
@@ -0,0 +1,108 @@
+package endtoend_test
+
+import (
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/pkg/errors"
+	"github.com/stretchr/testify/require"
+
+	"github.com/kopia/kopia/internal/apiclient"
+	"github.com/kopia/kopia/internal/serverapi"
+	"github.com/kopia/kopia/internal/testlogging"
+	"github.com/kopia/kopia/internal/testutil"
+	"github.com/kopia/kopia/snapshot/policy"
+	"github.com/kopia/kopia/tests/testenv"
+)
+
+// Verify that the "diagnostic/log" blobs are uploaded to the  repository when
+// the server exits.
+// Approach / steps:
+// - initialize a repo, note this uploads logs to the repo
+// - start the server
+// - create a "snapshot source" on the server via the server-control API
+// - remove all log blobs from the repo and check that there are 0.
+// - stop the server
+// - check whether or not the server uploaded the logs.
+func TestServerRepoLogsUploadedOnShutdown(t *testing.T) {
+	t.Parallel()
+
+	ctx := testlogging.Context(t)
+
+	runner := testenv.NewInProcRunner(t)
+	e := testenv.NewCLITest(t, testenv.RepoFormatNotImportant, runner)
+
+	e.RunAndExpectSuccess(t, "repo", "create", "filesystem", "--path", e.RepoDir, "--override-hostname=fake-hostname", "--override-username=fake-username")
+	defer e.RunAndExpectSuccess(t, "repo", "disconnect")
+
+	logs := e.RunAndExpectSuccess(t, "logs", "list")
+	require.Len(t, logs, 1, "repo create did not upload logs")
+
+	var sp testutil.ServerParameters
+
+	wait, _ := e.RunAndProcessStderr(t, sp.ProcessOutput,
+		"server", "start",
+		"--address=localhost:0",
+		"--insecure",
+		"--without-password",
+		"--tls-generate-rsa-key-size=2048", // use shorter key size to speed up generation,
+	)
+
+	require.NotEmpty(t, sp.BaseURL, "server base URL")
+
+	controlCli, err := apiclient.NewKopiaAPIClient(apiclient.Options{
+		BaseURL:  sp.BaseURL,
+		Username: defaultServerControlUsername,
+		Password: sp.ServerControlPassword,
+	})
+	require.NoError(t, err)
+
+	checkServerStartedOrFailed := func() bool {
+		var hs apiclient.HTTPStatusError
+
+		_, err := serverapi.Status(ctx, controlCli)
+
+		if errors.As(err, &hs) {
+			switch hs.HTTPStatusCode {
+			case http.StatusBadRequest:
+				return false
+			case http.StatusForbidden:
+				return false
+			}
+		}
+
+		return true
+	}
+
+	require.Eventually(t, checkServerStartedOrFailed, 10*time.Second, 100*time.Millisecond)
+	require.NoError(t, controlCli.FetchCSRFTokenForTesting(ctx))
+
+	keepDaily := policy.OptionalInt(3)
+
+	_, err = serverapi.CreateSnapshotSource(ctx, controlCli, &serverapi.CreateSnapshotSourceRequest{
+		Path: sharedTestDataDir1,
+		Policy: &policy.Policy{
+			RetentionPolicy: policy.RetentionPolicy{
+				KeepDaily: &keepDaily,
+			},
+		},
+		CreateSnapshot: false,
+	})
+
+	require.NoError(t, err)
+
+	lines := e.RunAndExpectSuccess(t, "server", "status", "--address", sp.BaseURL, "--server-control-password", sp.ServerControlPassword)
+	t.Logf("lines: %v", lines)
+
+	e.RunAndExpectSuccess(t, "logs", "cleanup", "--max-age=1ns")
+	logs = e.RunAndExpectSuccess(t, "logs", "list")
+	require.Empty(t, logs, "new logs were uploaded unexpectedly:", logs)
+
+	require.NoError(t, serverapi.Shutdown(ctx, controlCli))
+	require.NoError(t, wait())
+
+	logs = e.RunAndExpectSuccess(t, "logs", "list")
+
+	require.NotEmpty(t, logs, "server logs were not uploaded")
+}