Add checkpoint read and write for MySQL storage

cmd/example-mysql/main.go

@@ -16,6 +16,7 @@
 package main
 
 import (
+	"context"
 	"crypto/sha256"
 	"database/sql"
 	"flag"
@@ -39,6 +40,7 @@ var (
 func main() {
 	klog.InitFlags(nil)
 	flag.Parse()
+	ctx := context.Background()
 
 	db, err := sql.Open("mysql", *mysqlURI)
 	if err != nil {
@@ -48,11 +50,25 @@ func main() {
 	db.SetMaxOpenConns(*dbMaxOpenConns)
 	db.SetMaxIdleConns(*dbMaxIdleConns)
 
-	_, err = mysql.New(db)
+	storage, err := mysql.New(ctx, db)
 	if err != nil {
 		klog.Exitf("Failed to create new MySQL storage: %v", err)
 	}
 
+	http.HandleFunc("GET /checkpoint", func(w http.ResponseWriter, r *http.Request) {
+		checkpoint, err := storage.ReadCheckpoint(r.Context())
+		if err != nil {
+			klog.Errorf("/checkpoint: %v", err)
+			w.WriteHeader(http.StatusInternalServerError)
+			return
+		}
+
+		if _, err := w.Write(checkpoint); err != nil {
+			klog.Error(err)
+			return
+		}
+	})
+
 	http.HandleFunc("POST /add", func(w http.ResponseWriter, r *http.Request) {
 		b, err := io.ReadAll(r.Body)
 		if err != nil {

storage/mysql/mysql.go

@@ -16,19 +16,27 @@
 package mysql
 
 import (
+	"context"
 	"database/sql"
 
 	_ "github.com/go-sql-driver/mysql"
 	"k8s.io/klog/v2"
 )
 
+const (
+	selectCheckpointByIDSQL = "SELECT note FROM Checkpoint WHERE id = ?"
+	replaceCheckpointSQL    = "REPLACE INTO Checkpoint (id, note) VALUES (?, ?)"
+
+	checkpointID = 0
+)
+
 // Storage is a MySQL-based storage implementation for Tessera.
 type Storage struct {
 	db *sql.DB
 }
 
 // New creates a new instance of the MySQL-based Storage.
-func New(db *sql.DB) (*Storage, error) {
+func New(ctx context.Context, db *sql.DB) (*Storage, error) {
 	s := &Storage{
 		db: db,
 	}
@@ -37,5 +45,53 @@ func New(db *sql.DB) (*Storage, error) {
 		return nil, err
 	}
 
+	// Initialize checkpoint if there is no row in the Checkpoint table.
+	if _, err := s.ReadCheckpoint(ctx); err != nil {
+		if err != sql.ErrNoRows {
+			klog.Errorf("Failed to read checkpoint: %v", err)
+			return nil, err
+		}
+
+		klog.Infof("Initializing checkpoint")
+		if err := s.writeCheckpoint(ctx, []byte("")); err != nil {
+			klog.Errorf("Failed to write initial checkpoint: %v", err)
+			return nil, err
+		}
+	}
+
 	return s, nil
 }
+
+// ReadCheckpoint returns the latest stored checkpoint.
+func (s *Storage) ReadCheckpoint(ctx context.Context) ([]byte, error) {
+	row := s.db.QueryRowContext(ctx, selectCheckpointByIDSQL, checkpointID)
+	if err := row.Err(); err != nil {
+		return nil, err
+	}
+
+	var checkpoint []byte
+	return checkpoint, row.Scan(&checkpoint)
+}
+
+// writeCheckpoint stores a raw log checkpoint.
+func (s *Storage) writeCheckpoint(ctx context.Context, rawCheckpoint []byte) error {
+	// Get a Tx for making transaction requests.
+	tx, err := s.db.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+	// Defer a rollback in case anything fails.
+	defer func() {
+		if err := tx.Rollback(); err != nil && err != sql.ErrTxDone {
+			klog.Errorf("Failed to rollback in writeCheckpoint: %v", err)
+		}
+	}()
+
+	if _, err := tx.ExecContext(ctx, replaceCheckpointSQL, checkpointID, rawCheckpoint); err != nil {
+		klog.Errorf("Failed to execute replaceCheckpointSQL: %v", err)
+		return err
+	}
+
+	// Commit the transaction.
+	return tx.Commit()
+}

storage/mysql/schema.sql

@@ -0,0 +1,24 @@
+-- Copyright 2024 Google LLC
+--
+-- Licensed under the Apache License, Version 2.0 (the "License");
+-- you may not use this file except in compliance with the License.
+-- You may obtain a copy of the License at
+--
+--     http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing, software
+-- distributed under the License is distributed on an "AS IS" BASIS,
+-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+-- See the License for the specific language governing permissions and
+-- limitations under the License.
+
+-- MySQL version of the Trillian Tessera database schema.
+
+-- "Checkpoint" table stores a single row that records the current state of the log. It is updated after every sequence and integration.
+CREATE TABLE IF NOT EXISTS `Checkpoint` (
+  -- id is expected to be always 0 to maintain a maximum of a single row.
+  `id`    INT UNSIGNED NOT NULL,
+  -- note is the text signed by one or more keys in the checkpoint format. See https://github.com/C2SP/C2SP/blob/main/tlog-checkpoint.md and https://github.com/C2SP/C2SP/blob/main/signed-note.md.
+  `note`  MEDIUMBLOB NOT NULL,
+  PRIMARY KEY(`id`)
+);