repo: limit the duration of kopia.repository caching to 15 minutes (#1196)

added flags to specify kopia.repository cache duration

cli/command_repository_connect.go

@@ -54,6 +54,9 @@ type connectOptions struct {
 	connectReadonly               bool
 	connectDescription            string
 	connectEnableActions          bool
+
+	formatBlobCacheDuration time.Duration
+	disableFormatBlobCache  bool
 }
 
 func (c *connectOptions) setup(cmd *kingpin.CmdClause) {
@@ -69,6 +72,16 @@ func (c *connectOptions) setup(cmd *kingpin.CmdClause) {
 	cmd.Flag("readonly", "Make repository read-only to avoid accidental changes").BoolVar(&c.connectReadonly)
 	cmd.Flag("description", "Human-readable description of the repository").StringVar(&c.connectDescription)
 	cmd.Flag("enable-actions", "Allow snapshot actions").BoolVar(&c.connectEnableActions)
+	cmd.Flag("repository-format-cache-duration", "Duration of kopia.repository format blob cache").Hidden().DurationVar(&c.formatBlobCacheDuration)
+	cmd.Flag("disable-repository-format-cache", "Disable caching of kopia.repository format blob").Hidden().BoolVar(&c.disableFormatBlobCache)
+}
+
+func (c *connectOptions) getFormatBlobCacheDuration() time.Duration {
+	if c.disableFormatBlobCache {
+		return -1
+	}
+
+	return c.formatBlobCacheDuration
 }
 
 func (c *connectOptions) toRepoConnectOptions() *repo.ConnectOptions {
@@ -80,11 +93,12 @@ func (c *connectOptions) toRepoConnectOptions() *repo.ConnectOptions {
 			MaxListCacheDurationSec:   int(c.connectMaxListCacheDuration.Seconds()),
 		},
 		ClientOptions: repo.ClientOptions{
-			Hostname:      c.connectHostname,
-			Username:      c.connectUsername,
-			ReadOnly:      c.connectReadonly,
-			Description:   c.connectDescription,
-			EnableActions: c.connectEnableActions,
+			Hostname:                c.connectHostname,
+			Username:                c.connectUsername,
+			ReadOnly:                c.connectReadonly,
+			Description:             c.connectDescription,
+			EnableActions:           c.connectEnableActions,
+			FormatBlobCacheDuration: c.getFormatBlobCacheDuration(),
 		},
 	}
 }

cli/command_repository_set_client.go

@@ -2,6 +2,7 @@
 
 import (
 	"context"
+	"time"
 
 	"github.com/pkg/errors"
 
@@ -15,6 +16,9 @@ type commandRepositorySetClient struct {
 	repoClientOptionsUsername    []string
 	repoClientOptionsHostname    []string
 
+	formatBlobCacheDuration time.Duration
+	disableFormatBlobCache  bool
+
 	svc appServices
 }
 
@@ -26,6 +30,8 @@ func (c *commandRepositorySetClient) setup(svc appServices, parent commandParent
 	cmd.Flag("description", "Change description").StringsVar(&c.repoClientOptionsDescription)
 	cmd.Flag("username", "Change username").StringsVar(&c.repoClientOptionsUsername)
 	cmd.Flag("hostname", "Change hostname").StringsVar(&c.repoClientOptionsHostname)
+	cmd.Flag("repository-format-cache-duration", "Duration of kopia.repository format blob cache").DurationVar(&c.formatBlobCacheDuration)
+	cmd.Flag("disable-repository-format-cache", "Disable caching of kopia.repository format blob").BoolVar(&c.disableFormatBlobCache)
 	cmd.Action(svc.repositoryReaderAction(c.run))
 
 	c.svc = svc
@@ -79,6 +85,20 @@ func (c *commandRepositorySetClient) run(ctx context.Context, rep repo.Repositor
 		log(ctx).Infof("Setting local hostname to %v", opt.Hostname)
 	}
 
+	if v := c.formatBlobCacheDuration; v != 0 {
+		opt.FormatBlobCacheDuration = v
+		anyChange = true
+
+		log(ctx).Infof("Setting format blob cache duration to %v", v)
+	}
+
+	if c.disableFormatBlobCache {
+		opt.FormatBlobCacheDuration = -1
+		anyChange = true
+
+		log(ctx).Infof("Disabling format blob cache")
+	}
+
 	if !anyChange {
 		return errors.Errorf("no changes")
 	}

cli/command_repository_status.go

@@ -40,6 +40,12 @@ func (c *commandRepositoryStatus) run(ctx context.Context, rep repo.Repository)
 	c.out.printStdout("Username:            %v\n", rep.ClientOptions().Username)
 	c.out.printStdout("Read-only:           %v\n", rep.ClientOptions().ReadOnly)
 
+	if t := rep.ClientOptions().FormatBlobCacheDuration; t > 0 {
+		c.out.printStdout("Format blob cache:   %v\n", t)
+	} else {
+		c.out.printStdout("Format blob cache:   disabled\n")
+	}
+
 	dr, ok := rep.(repo.DirectRepository)
 	if !ok {
 		return nil

repo/local_config.go

@@ -6,6 +6,7 @@
 	"encoding/json"
 	"os"
 	"path/filepath"
+	"time"
 
 	"github.com/pkg/errors"
 
@@ -28,6 +29,8 @@ type ClientOptions struct {
 	Description string `json:"description,omitempty"`
 
 	EnableActions bool `json:"enableActions"`
+
+	FormatBlobCacheDuration time.Duration `json:"formatBlobCacheDuration,omitempty"`
 }
 
 // ApplyDefaults returns a copy of ClientOptions with defaults filled out.
@@ -44,6 +47,10 @@ func (o ClientOptions) ApplyDefaults(ctx context.Context, defaultDesc string) Cl
 		o.Description = defaultDesc
 	}
 
+	if o.FormatBlobCacheDuration == 0 {
+		o.FormatBlobCacheDuration = defaultFormatBlobCacheDuration
+	}
+
 	return o
 }
 

repo/open.go

@@ -13,6 +13,7 @@
 
 	"github.com/kopia/kopia/internal/atomicfile"
 	"github.com/kopia/kopia/internal/cache"
+	"github.com/kopia/kopia/internal/clock"
 	"github.com/kopia/kopia/repo/blob"
 	loggingwrapper "github.com/kopia/kopia/repo/blob/logging"
 	"github.com/kopia/kopia/repo/blob/readonly"
@@ -32,6 +33,10 @@
 // refresh indexes every 15 minutes while the repository remains open.
 const backgroundRefreshInterval = 15 * time.Minute
 
+// defaultFormatBlobCacheDuration is the duration for which we treat cached kopia.repository
+// as valid.
+const defaultFormatBlobCacheDuration = 15 * time.Minute
+
 const cacheDirMarkerContents = CacheDirMarkerHeader + `
 #
 # This file is a cache directory tag created by Kopia - Fast And Secure Open-Source Backup.
@@ -162,7 +167,7 @@ func openWithConfig(ctx context.Context, st blob.Storage, lc *LocalConfig, passw
 	caching = caching.CloneOrDefault()
 
 	// Read format blob, potentially from cache.
-	fb, err := readAndCacheFormatBlobBytes(ctx, st, caching.CacheDirectory)
+	fb, err := readAndCacheFormatBlobBytes(ctx, st, caching.CacheDirectory, lc.FormatBlobCacheDuration)
 	if err != nil {
 		return nil, errors.Wrap(err, "unable to read format blob")
 	}
@@ -280,19 +285,55 @@ func writeCacheMarker(cacheDir string) error {
 	return errors.Wrap(f.Close(), "error closing cache marker file")
 }
 
-func readAndCacheFormatBlobBytes(ctx context.Context, st blob.Storage, cacheDirectory string) ([]byte, error) {
+func formatBytesCachingEnabled(cacheDirectory string, validDuration time.Duration) bool {
+	if cacheDirectory == "" {
+		return false
+	}
+
+	return validDuration > 0
+}
+
+func readFormatBlobBytesFromCache(ctx context.Context, cachedFile string, validDuration time.Duration) ([]byte, error) {
+	if err := os.MkdirAll(filepath.Dir(cachedFile), cache.DirMode); err != nil && !os.IsExist(err) {
+		log(ctx).Errorf("unable to create cache directory: %v", err)
+	}
+
+	cst, err := os.Stat(cachedFile)
+	if err != nil {
+		return nil, errors.Wrap(err, "unable to open cache file")
+	}
+
+	if clock.Since(cst.ModTime()) > validDuration {
+		// got cached file, but it's too old, remove it
+		if err := os.Remove(cachedFile); err != nil {
+			log(ctx).Debugf("unable to remove cache file: %v", err)
+		}
+
+		return nil, errors.Errorf("cached file too old")
+	}
+
+	return ioutil.ReadFile(cachedFile) //nolint:gosec,wrapcheck
+}
+
+func readAndCacheFormatBlobBytes(ctx context.Context, st blob.Storage, cacheDirectory string, validDuration time.Duration) ([]byte, error) {
 	cachedFile := filepath.Join(cacheDirectory, "kopia.repository")
 
-	if cacheDirectory != "" {
-		if err := os.MkdirAll(cacheDirectory, cache.DirMode); err != nil && !os.IsExist(err) {
-			log(ctx).Errorf("unable to create cache directory: %v", err)
-		}
+	if validDuration == 0 {
+		validDuration = defaultFormatBlobCacheDuration
+	}
 
-		b, err := ioutil.ReadFile(cachedFile) //nolint:gosec
+	cacheEnabled := formatBytesCachingEnabled(cacheDirectory, validDuration)
+	if cacheEnabled {
+		b, err := readFormatBlobBytesFromCache(ctx, cachedFile, validDuration)
 		if err == nil {
-			// read from cache.
+			log(ctx).Debugf("kopia.repository retrieved from cache")
+
 			return b, nil
 		}
+
+		log(ctx).Debugf("kopia.repository could not be fetched from cache: %v", err)
+	} else {
+		log(ctx).Debugf("kopia.repository cache not enabled")
 	}
 
 	b, err := st.GetBlob(ctx, FormatBlobID, 0, -1)
@@ -300,7 +341,7 @@ func readAndCacheFormatBlobBytes(ctx context.Context, st blob.Storage, cacheDire
 		return nil, errors.Wrap(err, "error getting format blob")
 	}
 
-	if cacheDirectory != "" {
+	if cacheEnabled {
 		if err := atomicfile.Write(cachedFile, bytes.NewReader(b)); err != nil {
 			log(ctx).Errorf("warning: unable to write cache: %v", err)
 		}

tests/end_to_end_test/repository_set_client_test.go

@@ -15,7 +15,11 @@ func TestRepositorySetClient(t *testing.T) {
 
 	defer e.RunAndExpectSuccess(t, "repo", "disconnect")
 
-	e.RunAndExpectSuccess(t, "repo", "create", "filesystem", "--path", e.RepoDir, "--description", "My Repo", "--override-username", "myuser", "--override-hostname", "myhost")
+	e.RunAndExpectSuccess(t, "repo", "create",
+		"filesystem", "--path", e.RepoDir, "--description", "My Repo",
+		"--override-username", "myuser",
+		"--override-hostname", "myhost",
+		"--repository-format-cache-duration=7m")
 
 	sl := e.RunAndExpectSuccess(t, "repo", "status")
 	verifyHasLine(t, sl, func(l string) bool {
@@ -30,11 +34,16 @@ func TestRepositorySetClient(t *testing.T) {
 	verifyHasLine(t, sl, func(l string) bool {
 		return strings.Contains(l, "Hostname:") && strings.Contains(l, "myhost")
 	})
+	verifyHasLine(t, sl, func(l string) bool {
+		return strings.Contains(l, "Format blob cache:") && strings.Contains(l, "7m0s")
+	})
 
 	e.RunAndExpectSuccess(t, "repo", "set-client",
 		"--read-only",
 		"--description", "My Updated Repo",
-		"--hostname", "my-updated-host")
+		"--hostname", "my-updated-host",
+		"--disable-repository-format-cache",
+	)
 
 	sl = e.RunAndExpectSuccess(t, "repo", "status")
 	verifyHasLine(t, sl, func(l string) bool {
@@ -46,13 +55,21 @@ func TestRepositorySetClient(t *testing.T) {
 	verifyHasLine(t, sl, func(l string) bool {
 		return strings.Contains(l, "Hostname:") && strings.Contains(l, "my-updated-host")
 	})
+	verifyHasLine(t, sl, func(l string) bool {
+		return strings.Contains(l, "Format blob cache:") && strings.Contains(l, "disabled")
+	})
 
 	// repo is read-only
 	e.RunAndExpectFailure(t, "snapshot", "create", sharedTestDataDir1)
 
 	// set to read-write and snapshot will now succeeded
-	e.RunAndExpectSuccess(t, "repo", "set-client", "--read-write")
+	e.RunAndExpectSuccess(t, "repo", "set-client", "--read-write", "--repository-format-cache-duration=5s")
 	e.RunAndExpectSuccess(t, "snapshot", "create", sharedTestDataDir1)
+
+	sl = e.RunAndExpectSuccess(t, "repo", "status")
+	verifyHasLine(t, sl, func(l string) bool {
+		return strings.Contains(l, "Format blob cache:") && strings.Contains(l, "5s")
+	})
 }
 
 func verifyHasLine(t *testing.T, lines []string, ok func(s string) bool) {