Add conditional writes for the checkpoint.

experimental/gcp-log/function.go

@@ -323,7 +323,9 @@ func Integrate(w http.ResponseWriter, r *http.Request) {
 	return
 }
 
-func signAndWrite(ctx context.Context, cp *fmtlog.Checkpoint, cpNote note.Note, s note.Signer, client *storage.Client, origin string) error {
+// signAndWrite signs a checkpoint and writes the new checkpoint to GCS.
+func signAndWrite(ctx context.Context, cp *fmtlog.Checkpoint, cpNote note.Note,
+	s note.Signer, client *storage.Client, origin string) error {
 	cp.Origin = origin
 	cpNote.Text = string(cp.Marshal())
 	cpNoteSigned, err := note.Sign(&cpNote, s)

experimental/gcp-log/internal/storage/storage.go

@@ -54,6 +54,9 @@ type Client struct {
 	// Note that nextSeq may be <= than the actual next available number, but
 	// never greater.
 	nextSeq uint64
+	// checkpointGen is the GCS object generation number that this client last
+	// read. This is useful for read-modify-write operation of the checkpoint.
+	checkpointGen int64
 }
 
 // NewClient returns a Client which allows interaction with the log stored in
@@ -65,9 +68,10 @@ func NewClient(ctx context.Context, projectID, bucket string) (*Client, error) {
 	}
 
 	return &Client{
-		gcsClient: c,
-		projectID: projectID,
-		bucket:    bucket,
+		gcsClient:     c,
+		projectID:     projectID,
+		bucket:        bucket,
+		checkpointGen: 0,
 	}, nil
 }
 
@@ -116,14 +120,17 @@ func (c *Client) SetNextSeq(num uint64) {
 	c.nextSeq = num
 }
 
-// WriteCheckpoint stores a raw log checkpoint on GCS.
+// WriteCheckpoint stores a raw log checkpoint on GCS if it matches the
+// generation that the client thinks the checkpoint is.
 func (c *Client) WriteCheckpoint(ctx context.Context, newCPRaw []byte) error {
 	bkt := c.gcsClient.Bucket(c.bucket)
 	obj := bkt.Object(layout.CheckpointPath)
-	w := obj.NewWriter(ctx)
+
+	w := obj.If(gcs.Conditions{GenerationMatch: c.checkpointGen}).NewWriter(ctx)
 	if _, err := w.Write(newCPRaw); err != nil {
 		return err
 	}
+	c.checkpointGen++
 	return w.Close()
 }
 
@@ -132,13 +139,22 @@ func (c *Client) ReadCheckpoint(ctx context.Context) ([]byte, error) {
 	bkt := c.gcsClient.Bucket(c.bucket)
 	obj := bkt.Object(layout.CheckpointPath)
 
+	// Get the GCS generation number.
+	attrs, err := obj.Attrs(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("Object(%q).Attrs: %w", obj, err)
+	}
+	c.checkpointGen = attrs.Generation
+
+	// Get the content of the checkpoint.
 	r, err := obj.NewReader(ctx)
 	if err != nil {
 		return nil, err
 	}
 	defer r.Close()
 
-	return io.ReadAll(r)
+	content, err := io.ReadAll(r)
+	return content, err
 }
 
 // GetTile returns the tile at the given tile-level and tile-index.