package mysql
import (
"context"
"database/sql"
"errors"
"fmt"
"time"
"github.com/google/trillian"
"google.golang.org/protobuf/types/known/timestamppb"
"k8s.io/klog/v2"
)
const (
selectQueuedLeavesSQL = `SELECT LeafIdentityHash,MerkleLeafHash,QueueTimestampNanos
FROM Unsequenced
WHERE TreeID=?
AND Bucket=0
AND QueueTimestampNanos<=?
ORDER BY QueueTimestampNanos,LeafIdentityHash ASC LIMIT ?`
insertUnsequencedEntrySQL = `INSERT INTO Unsequenced(TreeId,Bucket,LeafIdentityHash,MerkleLeafHash,QueueTimestampNanos)
VALUES(?,0,?,?,?)`
deleteUnsequencedSQL = "DELETE FROM Unsequenced WHERE TreeId=? AND Bucket=0 AND QueueTimestampNanos=? AND LeafIdentityHash=?"
)
type dequeuedLeaf struct {
queueTimestampNanos int64
leafIdentityHash []byte
}
func dequeueInfo(leafIDHash []byte, queueTimestamp int64) dequeuedLeaf {
return dequeuedLeaf{queueTimestampNanos: queueTimestamp, leafIdentityHash: leafIDHash}
}
func (t *logTreeTX) dequeueLeaf(rows *sql.Rows) (*trillian.LogLeaf, dequeuedLeaf, error) {
var leafIDHash []byte
var merkleHash []byte
var queueTimestamp int64
err := rows.Scan(&leafIDHash, &merkleHash, &queueTimestamp)
if err != nil {
klog.Warningf("Error scanning work rows: %s", err)
return nil, dequeuedLeaf{}, err
}
queueTimestampProto := timestamppb.New(time.Unix(0, queueTimestamp))
if err := queueTimestampProto.CheckValid(); err != nil {
return nil, dequeuedLeaf{}, fmt.Errorf("got invalid queue timestamp: %w", err)
}
leaf := &trillian.LogLeaf{
LeafIdentityHash: leafIDHash,
MerkleLeafHash: merkleHash,
QueueTimestamp: queueTimestampProto,
}
return leaf, dequeueInfo(leafIDHash, queueTimestamp), nil
}
func queueArgs(_ int64, _ []byte, queueTimestamp time.Time) []interface{} {
return []interface{}{queueTimestamp.UnixNano()}
}
func (t *logTreeTX) UpdateSequencedLeaves(ctx context.Context, leaves []*trillian.LogLeaf) error {
dequeuedLeaves := make([]dequeuedLeaf, 0, len(leaves))
for _, leaf := range leaves {
if len(leaf.LeafIdentityHash) != t.hashSizeBytes {
return errors.New("sequenced leaf has incorrect hash size")
}
if err := leaf.IntegrateTimestamp.CheckValid(); err != nil {
return fmt.Errorf("got invalid integrate timestamp: %w", err)
}
iTimestamp := leaf.IntegrateTimestamp.AsTime()
_, err := t.tx.ExecContext(
ctx,
insertSequencedLeafSQL+valuesPlaceholder5,
t.treeID,
leaf.LeafIdentityHash,
leaf.MerkleLeafHash,
leaf.LeafIndex,
iTimestamp.UnixNano())
if err != nil {
klog.Warningf("Failed to update sequenced leaves: %s", err)
return err
}
qe, ok := t.dequeued[string(leaf.LeafIdentityHash)]
if !ok {
return fmt.Errorf("attempting to update leaf that wasn't dequeued. IdentityHash: %x", leaf.LeafIdentityHash)
}
dequeuedLeaves = append(dequeuedLeaves, qe)
}
return t.removeSequencedLeaves(ctx, dequeuedLeaves)
}
func (t *logTreeTX) removeSequencedLeaves(ctx context.Context, leaves []dequeuedLeaf) error {
start := time.Now()
stx, err := t.tx.PrepareContext(ctx, deleteUnsequencedSQL)
if err != nil {
klog.Warningf("Failed to prep delete statement for sequenced work: %v", err)
return err
}
defer func() {
if err := stx.Close(); err != nil {
klog.Errorf("stx.Close(): %v", err)
}
}()
for _, dql := range leaves {
result, err := stx.ExecContext(ctx, t.treeID, dql.queueTimestampNanos, dql.leafIdentityHash)
err = checkResultOkAndRowCountIs(result, err, int64(1))
if err != nil {
return err
}
}
observe(dequeueRemoveLatency, time.Since(start), labelForTX(t))
return nil
}