351 lines
13 KiB
Go
351 lines
13 KiB
Go
package store
|
|
|
|
import (
|
|
"context"
|
|
"crypto/subtle"
|
|
"database/sql"
|
|
"errors"
|
|
"fmt"
|
|
"io/fs"
|
|
"math"
|
|
"path/filepath"
|
|
|
|
rvboxv1 "github.com/rvbox/rvbox/gen/go/rvbox/v1"
|
|
"github.com/rvbox/rvbox/internal/domain"
|
|
"google.golang.org/protobuf/proto"
|
|
)
|
|
|
|
const (
|
|
FaultAfterSegmentSyncBeforeMetadata = "after_segment_sync_before_metadata"
|
|
FaultAfterMetadataCommit = "after_metadata_commit"
|
|
)
|
|
|
|
var (
|
|
ErrStoreClosed = errors.New("store is closed")
|
|
ErrCommandNotFound = errors.New("command not found")
|
|
ErrEventConflict = errors.New("event sequence already has different immutable content")
|
|
ErrEventSequenceGap = errors.New("event sequence is not the next expected value")
|
|
)
|
|
|
|
type FaultInjector interface {
|
|
Checkpoint(name string) error
|
|
}
|
|
|
|
type FaultInjectorFunc func(name string) error
|
|
|
|
func (function FaultInjectorFunc) Checkpoint(name string) error { return function(name) }
|
|
|
|
type EventAppend struct {
|
|
IssueUUID [16]byte
|
|
ClientID string
|
|
SessionGeneration uint64
|
|
EventSeq uint64
|
|
ObservedUnixNano int64
|
|
ReceiptUnixNano int64
|
|
EventType uint16
|
|
Stream uint16
|
|
Compression uint16
|
|
RawLength uint64
|
|
Payload []byte
|
|
ImmutableSHA256 [32]byte
|
|
Lifecycle *rvboxv1.CommandLifecycle
|
|
LifecycleRevision uint64
|
|
Output bool
|
|
UseCloseout bool
|
|
}
|
|
|
|
type EventAppendResult struct {
|
|
ThroughEventSeq uint64
|
|
Duplicate bool
|
|
}
|
|
|
|
type activeSegment struct {
|
|
segment *Segment
|
|
ordinal uint32
|
|
committed uint64
|
|
}
|
|
|
|
func (store *Store) AppendCommandEvent(ctx context.Context, event EventAppend) (EventAppendResult, error) {
|
|
var result EventAppendResult
|
|
if isZeroUUID(event.IssueUUID) || event.EventSeq == 0 || event.EventSeq > math.MaxInt64 || event.EventType == 0 || event.ReceiptUnixNano <= 0 || allZero(event.ImmutableSHA256[:]) {
|
|
return result, ErrInvalidSegmentRecord
|
|
}
|
|
record := SegmentRecord{
|
|
OwnerUUID: event.IssueUUID, Sequence: event.EventSeq, ObservedUnixNano: event.ObservedUnixNano,
|
|
ReceiptUnixNano: event.ReceiptUnixNano, Kind: PayloadKindCommandEvent, Stream: event.Stream,
|
|
Compression: event.Compression, RawLength: event.RawLength, Payload: event.Payload,
|
|
}
|
|
encoded, err := EncodeSegmentRecord(record, DefaultSegmentLimits())
|
|
if err != nil {
|
|
return result, err
|
|
}
|
|
|
|
store.writeMu.Lock()
|
|
defer store.writeMu.Unlock()
|
|
database, err := store.openDatabase()
|
|
if err != nil {
|
|
return result, err
|
|
}
|
|
var lastSequence, commandOutputCharged, commandCharged, clientCharged, serverCharged, closeoutRemaining uint64
|
|
var currentLifecycle uint32
|
|
var currentRevision uint64
|
|
var clientID string
|
|
query := `SELECT commands.last_event_seq, commands.output_charged_bytes,
|
|
commands.charged_bytes, commands.closeout_remaining_bytes, commands.client_id, clients.charged_bytes,
|
|
storage_counters.command_charged_bytes, commands.lifecycle, commands.revision
|
|
FROM commands JOIN clients ON clients.client_id = commands.client_id
|
|
JOIN storage_counters ON storage_counters.singleton = 1 WHERE commands.issue_uuid = ?`
|
|
arguments := []any{event.IssueUUID[:]}
|
|
if event.ClientID != "" {
|
|
query += ` AND commands.client_id = ?`
|
|
arguments = append(arguments, event.ClientID)
|
|
}
|
|
if event.SessionGeneration != 0 {
|
|
query += ` AND commands.target_session_generation = ?`
|
|
arguments = append(arguments, event.SessionGeneration)
|
|
}
|
|
err = database.QueryRowContext(ctx, query, arguments...).Scan(
|
|
&lastSequence, &commandOutputCharged, &commandCharged, &closeoutRemaining, &clientID, &clientCharged, &serverCharged, ¤tLifecycle, ¤tRevision)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return result, ErrCommandNotFound
|
|
}
|
|
if err != nil {
|
|
return result, err
|
|
}
|
|
if event.EventSeq <= lastSequence {
|
|
var storedHash []byte
|
|
if err := database.QueryRowContext(ctx, `SELECT immutable_sha256 FROM command_events WHERE issue_uuid = ? AND event_seq = ?`, event.IssueUUID[:], event.EventSeq).Scan(&storedHash); err != nil {
|
|
return result, fmt.Errorf("load duplicate event: %w", err)
|
|
}
|
|
if len(storedHash) != len(event.ImmutableSHA256) || subtle.ConstantTimeCompare(storedHash, event.ImmutableSHA256[:]) != 1 {
|
|
return result, ErrEventConflict
|
|
}
|
|
return EventAppendResult{ThroughEventSeq: lastSequence, Duplicate: true}, nil
|
|
}
|
|
if event.EventSeq != lastSequence+1 {
|
|
return result, ErrEventSequenceGap
|
|
}
|
|
if event.Lifecycle != nil {
|
|
if event.LifecycleRevision == 0 || event.LifecycleRevision != currentRevision {
|
|
return result, errors.New("lifecycle event revision does not match command")
|
|
}
|
|
if !domain.CanTransition(rvboxv1.CommandLifecycle(currentLifecycle), *event.Lifecycle) {
|
|
return result, domain.ValidateTransition(rvboxv1.CommandLifecycle(currentLifecycle), *event.Lifecycle)
|
|
}
|
|
}
|
|
rowsCharged, indexesCharged := uint64(1), uint64(1)
|
|
if store.willCreateSegment(event.IssueUUID, uint64(len(encoded))) {
|
|
rowsCharged++
|
|
indexesCharged += 2
|
|
}
|
|
chargedBytes, err := EstimateCharge(ChargeInput{EncodedBytes: uint64(len(encoded)), SQLiteRows: rowsCharged, IndexEntries: indexesCharged})
|
|
if err != nil {
|
|
return result, err
|
|
}
|
|
freeBytes, err := store.freeSpaceProbe.AvailableBytes(store.dataDir)
|
|
if err != nil {
|
|
return result, fmt.Errorf("check filesystem capacity: %w", err)
|
|
}
|
|
reservation, err := CheckReservation(store.quotaLimits, ReservationState{
|
|
CommandOutputCharged: commandOutputCharged, CommandTotalCharged: commandCharged,
|
|
ClientTotalCharged: clientCharged, ServerTotalCharged: serverCharged,
|
|
CloseoutRemaining: closeoutRemaining, FilesystemFreeBytes: freeBytes,
|
|
}, ReservationRequest{
|
|
ChargedBytes: chargedBytes, PhysicalBytes: uint64(len(encoded)), Output: event.Output || event.Stream > 0,
|
|
UseCloseout: event.UseCloseout,
|
|
})
|
|
if err != nil {
|
|
return result, err
|
|
}
|
|
|
|
active, err := store.segmentForAppend(ctx, database, event.IssueUUID, uint64(len(encoded)))
|
|
if err != nil {
|
|
return result, err
|
|
}
|
|
recordOffset := active.segment.offset
|
|
committedEnd, err := active.segment.Append(record)
|
|
if err != nil {
|
|
store.invalidateActiveSegment(event.IssueUUID)
|
|
return result, err
|
|
}
|
|
if err := store.checkpoint(FaultAfterSegmentSyncBeforeMetadata); err != nil {
|
|
store.invalidateActiveSegment(event.IssueUUID)
|
|
return result, err
|
|
}
|
|
|
|
tx, err := database.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
store.invalidateActiveSegment(event.IssueUUID)
|
|
return result, err
|
|
}
|
|
streamMix := uint64(0)
|
|
if event.Stream > 0 {
|
|
streamMix = uint64(1) << (event.Stream - 1)
|
|
}
|
|
if active.committed == 0 {
|
|
_, err = tx.ExecContext(ctx, `INSERT INTO output_segments (
|
|
issue_uuid, ordinal, path, committed_end_offset, min_event_seq, max_event_seq, stream_mix,
|
|
compressed_bytes, raw_bytes, checksum, sealed, created_at
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, 0, ?)`, event.IssueUUID[:], active.ordinal, active.segment.Name(), committedEnd,
|
|
event.EventSeq, event.EventSeq, streamMix, len(event.Payload), event.RawLength, event.ReceiptUnixNano)
|
|
} else {
|
|
var update sql.Result
|
|
update, err = tx.ExecContext(ctx, `UPDATE output_segments SET
|
|
committed_end_offset = ?, max_event_seq = ?, stream_mix = stream_mix | ?,
|
|
compressed_bytes = compressed_bytes + ?, raw_bytes = raw_bytes + ?
|
|
WHERE issue_uuid = ? AND ordinal = ? AND committed_end_offset = ?`, committedEnd, event.EventSeq, streamMix,
|
|
len(event.Payload), event.RawLength, event.IssueUUID[:], active.ordinal, active.committed)
|
|
if err == nil {
|
|
var affected int64
|
|
affected, err = update.RowsAffected()
|
|
if err == nil && affected != 1 {
|
|
err = errors.New("segment committed offset changed unexpectedly")
|
|
}
|
|
}
|
|
}
|
|
if err == nil {
|
|
_, err = tx.ExecContext(ctx, `INSERT INTO command_events (
|
|
issue_uuid, event_seq, observed_at, server_receipt_time, event_type, compression, raw_bytes, stored_bytes,
|
|
payload, segment_ordinal, segment_record_offset, segment_record_length, immutable_sha256
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?)`, event.IssueUUID[:], event.EventSeq, event.ObservedUnixNano,
|
|
event.ReceiptUnixNano, event.EventType, event.Compression, event.RawLength, len(event.Payload), active.ordinal,
|
|
recordOffset, uint64(len(encoded)), event.ImmutableSHA256[:])
|
|
}
|
|
if err == nil {
|
|
var update sql.Result
|
|
commandUpdate := `UPDATE commands SET last_event_seq = ?,
|
|
retained_compressed_bytes = retained_compressed_bytes + ?, output_charged_bytes = ?, charged_bytes = ?,
|
|
closeout_remaining_bytes = ?`
|
|
commandArgs := []any{event.EventSeq, len(event.Payload), reservation.CommandOutputCharged, reservation.CommandTotalCharged, reservation.CloseoutRemaining}
|
|
if event.Lifecycle != nil {
|
|
terminal := 0
|
|
if domain.IsTerminal(*event.Lifecycle) {
|
|
terminal = 1
|
|
}
|
|
commandUpdate = `UPDATE commands SET lifecycle = ?, terminal_time = CASE WHEN ? = 1 THEN ? ELSE terminal_time END, ` + commandUpdate[len("UPDATE commands SET "):]
|
|
commandArgs = append([]any{uint32(*event.Lifecycle), terminal, event.ReceiptUnixNano}, commandArgs...)
|
|
}
|
|
commandUpdate += ` WHERE issue_uuid = ? AND last_event_seq = ?`
|
|
commandArgs = append(commandArgs, event.IssueUUID[:], lastSequence)
|
|
update, err = tx.ExecContext(ctx, commandUpdate, commandArgs...)
|
|
if err == nil {
|
|
var affected int64
|
|
affected, err = update.RowsAffected()
|
|
if err == nil && affected != 1 {
|
|
err = errors.New("command event sequence changed unexpectedly")
|
|
}
|
|
}
|
|
}
|
|
if err == nil && event.EventType == 7 {
|
|
// Stdin acknowledgements are command events, but their durable delivery
|
|
// cursor lives in stdin_writes. Mark the cumulative prefix in the same
|
|
// transaction as the event so a crash cannot make an acknowledged write
|
|
// replay forever or release it before its event is durable.
|
|
var wire rvboxv1.CommandEvent
|
|
if unmarshalErr := proto.Unmarshal(event.Payload, &wire); unmarshalErr != nil || wire.GetStdinAck() == nil || wire.GetStdinAck().GetWriteSeq() == 0 {
|
|
err = ErrInvalidSegmentRecord
|
|
} else {
|
|
_, err = tx.ExecContext(ctx, `UPDATE stdin_writes SET acknowledged = 1 WHERE issue_uuid = ? AND write_seq <= ?`, event.IssueUUID[:], wire.GetStdinAck().GetWriteSeq())
|
|
}
|
|
}
|
|
if err == nil {
|
|
var update sql.Result
|
|
update, err = tx.ExecContext(ctx, `UPDATE clients SET charged_bytes = ? WHERE client_id = ? AND charged_bytes = ?`, reservation.ClientTotalCharged, clientID, clientCharged)
|
|
if err == nil {
|
|
var affected int64
|
|
affected, err = update.RowsAffected()
|
|
if err == nil && affected != 1 {
|
|
err = errors.New("client quota counter changed unexpectedly")
|
|
}
|
|
}
|
|
}
|
|
if err == nil {
|
|
var update sql.Result
|
|
update, err = tx.ExecContext(ctx, `UPDATE storage_counters SET command_charged_bytes = ? WHERE singleton = 1 AND command_charged_bytes = ?`, reservation.ServerTotalCharged, serverCharged)
|
|
if err == nil {
|
|
var affected int64
|
|
affected, err = update.RowsAffected()
|
|
if err == nil && affected != 1 {
|
|
err = errors.New("server quota counter changed unexpectedly")
|
|
}
|
|
}
|
|
}
|
|
if err != nil {
|
|
_ = tx.Rollback()
|
|
store.invalidateActiveSegment(event.IssueUUID)
|
|
return result, fmt.Errorf("commit event metadata: %w", err)
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
store.invalidateActiveSegment(event.IssueUUID)
|
|
return result, fmt.Errorf("commit event metadata: %w", err)
|
|
}
|
|
active.committed = committedEnd
|
|
result.ThroughEventSeq = event.EventSeq
|
|
if err := store.checkpoint(FaultAfterMetadataCommit); err != nil {
|
|
return result, err
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func (store *Store) willCreateSegment(owner [16]byte, recordLength uint64) bool {
|
|
current := store.activeSegments[owner]
|
|
return current == nil || (current.segment.offset != 0 && current.segment.offset+recordLength > store.segmentTarget)
|
|
}
|
|
|
|
func (store *Store) segmentForAppend(ctx context.Context, database *sql.DB, owner [16]byte, recordLength uint64) (*activeSegment, error) {
|
|
if current := store.activeSegments[owner]; current != nil {
|
|
if current.segment.offset == 0 || current.segment.offset+recordLength <= store.segmentTarget {
|
|
return current, nil
|
|
}
|
|
if err := current.segment.Close(); err != nil {
|
|
return nil, err
|
|
}
|
|
delete(store.activeSegments, owner)
|
|
}
|
|
var highest int64
|
|
if err := database.QueryRowContext(ctx, `SELECT COALESCE(MAX(ordinal), -1) FROM output_segments WHERE issue_uuid = ?`, owner[:]).Scan(&highest); err != nil {
|
|
return nil, err
|
|
}
|
|
if highest >= math.MaxUint32 {
|
|
return nil, errors.New("segment ordinal exhausted")
|
|
}
|
|
ordinal := uint32(highest + 1)
|
|
for {
|
|
segment, err := CreateSegment(store.segmentDirectory(), owner, ordinal, DefaultSegmentLimits())
|
|
if err == nil {
|
|
active := &activeSegment{segment: segment, ordinal: ordinal}
|
|
store.activeSegments[owner] = active
|
|
return active, nil
|
|
}
|
|
if !errors.Is(err, fs.ErrExist) || ordinal == math.MaxUint32 {
|
|
return nil, err
|
|
}
|
|
ordinal++
|
|
}
|
|
}
|
|
|
|
func (store *Store) invalidateActiveSegment(owner [16]byte) {
|
|
if active := store.activeSegments[owner]; active != nil {
|
|
_ = active.segment.Close()
|
|
delete(store.activeSegments, owner)
|
|
}
|
|
}
|
|
|
|
func (store *Store) openDatabase() (*sql.DB, error) {
|
|
store.mu.Lock()
|
|
defer store.mu.Unlock()
|
|
if store.db == nil {
|
|
return nil, ErrStoreClosed
|
|
}
|
|
return store.db, nil
|
|
}
|
|
|
|
func (store *Store) checkpoint(name string) error {
|
|
if store.faultInjector == nil {
|
|
return nil
|
|
}
|
|
return store.faultInjector.Checkpoint(name)
|
|
}
|
|
|
|
func (store *Store) segmentDirectory() string { return filepath.Join(store.dataDir, "segments") }
|