feat: ACI-4298 Retry cache writes with offset (#211)

Author: zsolt-marta-bitrise

cmd/xcode/mocks/kv_storage.go

@@ -95,9 +95,9 @@ func (c *Client) DownloadStream(ctx context.Context, destination io.Writer, key
 
 	downloadErr := retry.Times(c.downloadRetry).Wait(c.downloadRetryWait).TryWithAbort(func(attempt uint) (error, bool) {
 		if attempt == 0 {
-			c.logger.Debugf("Downloading %s", key)
+			c.logger.TDebugf("Downloading %s", key)
 		} else {
-			c.logger.Infof("%d. attempt to download %s with offset %d", attempt+1, key, offset)
+			c.logger.TInfof("%d. attempt to download %s with offset %d", attempt+1, key, offset)
 		}
 
 		timeoutCtx, cancel := context.WithTimeout(ctx, 60*time.Second)
@@ -127,7 +127,6 @@ func (c *Client) DownloadStream(ctx context.Context, destination io.Writer, key
 			return fmt.Errorf("download archive: %w", err), false
 		}
 
-		c.logger.Debugf("Response metadata: %+v", kvReader.Metadata())
 		expectedHash = kvReader.Metadata()["x-flare-blob-validation-sha256"]
 		expectedHash = strings.TrimPrefix(expectedHash, "blob/")
 
@@ -143,7 +142,7 @@ func (c *Client) DownloadStream(ctx context.Context, destination io.Writer, key
 		if expectedHash != fileHash {
 			return fmt.Errorf("downloaded file hash mismatch: expected %s, got %s", expectedHash, fileHash)
 		} else {
-			c.logger.Debugf("Downloaded hash matches expected: %s", expectedHash)
+			c.logger.TDebugf("Downloaded %s hash matches expected: %s", key, expectedHash)
 		}
 	}
 

internal/build_cache/kv/download.go

@@ -19,9 +19,16 @@ import (
 )
 
 type PutParams struct {
-	Name      string
-	Sha256Sum string
-	FileSize  int64
+	Name            string
+	Sha256Sum       string
+	FileSize        int64
+	Offset          int64
+	DeleteOnRewrite bool
+}
+
+type WriteStatus struct {
+	Complete      bool
+	CommittedSize int64
 }
 
 type FileDigest struct {
@@ -73,6 +80,9 @@ func (c *Client) initiatePut(ctx context.Context, params PutParams) (*writer, er
 		"x-flare-blob-validation-level", "error",
 		"x-flare-no-skip-duplicate-writes", "true",
 	))
+	if params.DeleteOnRewrite {
+		md.Set("x-cache-delete-on-rewrite", "true")
+	}
 	// Timeout is the responsibility of the caller
 	ctx = metadata.NewOutgoingContext(ctx, md)
 
@@ -91,7 +101,7 @@ func (c *Client) initiatePut(ctx context.Context, params PutParams) (*writer, er
 	return &writer{
 		stream:       stream,
 		resourceName: resourceName,
-		offset:       0,
+		offset:       params.Offset,
 		fileSize:     params.FileSize,
 	}, nil
 }
@@ -311,3 +321,27 @@ func (c *Client) getMethodCallMetadata(logMD bool) metadata.MD {
 
 	return md
 }
+
+func (c *Client) QueryWriteStatus(ctx context.Context, name string) (WriteStatus, error) {
+	resourceName := fmt.Sprintf("kv/%s", name)
+
+	timeoutCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
+	defer cancel()
+	callCtx := metadata.NewOutgoingContext(timeoutCtx, c.getMethodCallMetadata(false))
+	resp, err := c.bitriseKVClient.WriteStatus(callCtx, &bytestream.QueryWriteStatusRequest{
+		ResourceName: resourceName,
+	})
+	if err != nil {
+		st, ok := status.FromError(err)
+		if ok && st.Code() == codes.Unauthenticated {
+			return WriteStatus{}, ErrCacheUnauthenticated
+		}
+
+		return WriteStatus{}, fmt.Errorf("query write status: %w", err)
+	}
+
+	return WriteStatus{
+		Complete:      resp.GetComplete(),
+		CommittedSize: resp.GetCommittedSize(),
+	}, nil
+}

internal/build_cache/kv/methods.go

@@ -23,6 +23,13 @@ func init() {
 	mockLogger.On("Infof", mock.Anything, mock.Anything).Return()
 	mockLogger.On("Infof", mock.Anything).Return()
 	mockLogger.On("Infof").Return()
+	mockLogger.On("TInfof", mock.Anything, mock.Anything, mock.Anything).Return()
+	mockLogger.On("TInfof", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return()
+	mockLogger.On("TInfof", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return()
+	mockLogger.On("TInfof", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return()
+	mockLogger.On("TInfof", mock.Anything, mock.Anything).Return()
+	mockLogger.On("TInfof", mock.Anything).Return()
+	mockLogger.On("TInfof").Return()
 	mockLogger.On("TDebugf", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return()
 	mockLogger.On("TDebugf", mock.Anything, mock.Anything).Return()
 	mockLogger.On("TDebugf", mock.Anything).Return()
@@ -39,4 +46,8 @@ func init() {
 	mockLogger.On("Warnf", mock.Anything, mock.Anything).Return()
 	mockLogger.On("Warnf", mock.Anything, mock.Anything, mock.Anything).Return()
 	mockLogger.On("Warnf", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return()
+	mockLogger.On("TWarnf", mock.Anything).Return()
+	mockLogger.On("TWarnf", mock.Anything, mock.Anything).Return()
+	mockLogger.On("TWarnf", mock.Anything, mock.Anything, mock.Anything).Return()
+	mockLogger.On("TWarnf", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return()
 }