422 lines
15 KiB
Go
422 lines
15 KiB
Go
package spool
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/rvbox/rvbox/internal/domain"
|
|
)
|
|
|
|
type Command struct {
|
|
IssueUUID domain.UUID
|
|
ImmutableSHA256 [32]byte
|
|
Revision uint64
|
|
Phase uint32
|
|
Terminal bool
|
|
}
|
|
|
|
type Acceptance struct {
|
|
Duplicate bool
|
|
Command Command
|
|
}
|
|
|
|
type EventInput struct {
|
|
Kind uint32
|
|
Compression uint32
|
|
RawBytes uint64
|
|
Payload []byte
|
|
Output bool
|
|
UseCloseout bool
|
|
CreatedAt time.Time
|
|
}
|
|
|
|
type Event struct {
|
|
IssueUUID domain.UUID
|
|
LocalOrdinal uint64
|
|
EventSeq uint64
|
|
Kind uint32
|
|
Compression uint32
|
|
RawBytes uint64
|
|
Payload []byte
|
|
CreatedAt time.Time
|
|
}
|
|
|
|
func (store *Store) AcceptCommand(ctx context.Context, command Command, acceptedAt time.Time) (Acceptance, error) {
|
|
if !validUUID(command.IssueUUID) || command.Revision == 0 || command.Phase == 0 || command.Phase > 11 || command.Terminal != isTerminalPhase(command.Phase) || acceptedAt.IsZero() {
|
|
return Acceptance{}, errors.New("invalid command acceptance")
|
|
}
|
|
tx, err := store.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return Acceptance{}, err
|
|
}
|
|
defer tx.Rollback()
|
|
existing, found, err := commandByUUID(ctx, tx, command.IssueUUID)
|
|
if err != nil {
|
|
return Acceptance{}, err
|
|
}
|
|
if found {
|
|
if existing.ImmutableSHA256 != command.ImmutableSHA256 {
|
|
return Acceptance{}, ErrCommandConflict
|
|
}
|
|
return Acceptance{Duplicate: true, Command: existing}, nil
|
|
}
|
|
var tombstoneHash []byte
|
|
err = tx.QueryRowContext(ctx, `SELECT immutable_sha256 FROM command_tombstones WHERE issue_uuid = ?`, command.IssueUUID[:]).Scan(&tombstoneHash)
|
|
if err == nil {
|
|
if string(tombstoneHash) != string(command.ImmutableSHA256[:]) {
|
|
return Acceptance{}, ErrCommandConflict
|
|
}
|
|
return Acceptance{}, ErrAlreadyExecuted
|
|
}
|
|
if err != sql.ErrNoRows {
|
|
return Acceptance{}, err
|
|
}
|
|
charge, err := EstimateCharge(ChargeInput{SQLiteRows: 1, IndexEntries: 2})
|
|
if err != nil {
|
|
return Acceptance{}, err
|
|
}
|
|
clientTotal, err := clientTotalCharge(ctx, tx)
|
|
if err != nil {
|
|
return Acceptance{}, err
|
|
}
|
|
decision, err := CheckReservation(store.quotaLimits, ReservationState{ClientTotalCharged: clientTotal, CloseoutRemaining: store.quotaLimits.CloseoutReserveBytes}, ReservationRequest{ChargedBytes: charge})
|
|
if err != nil {
|
|
return Acceptance{}, err
|
|
}
|
|
_, err = tx.ExecContext(ctx, `INSERT INTO commands(issue_uuid, immutable_sha256, command_revision, phase, terminal, total_charged_bytes, closeout_remaining_bytes, accepted_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, command.IssueUUID[:], command.ImmutableSHA256[:], command.Revision, command.Phase, boolInt(command.Terminal), decision.CommandTotalCharged, decision.CloseoutRemaining, acceptedAt.UnixNano())
|
|
if err != nil {
|
|
return Acceptance{}, err
|
|
}
|
|
if err := updateClientTotalCharge(ctx, tx, decision.ClientTotalCharged); err != nil {
|
|
return Acceptance{}, err
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return Acceptance{}, err
|
|
}
|
|
return Acceptance{Command: command}, nil
|
|
}
|
|
|
|
// AppendEvent assigns command-local order only. EventSeq remains zero until
|
|
// AssignSendWindow makes the durable event eligible for a wire send.
|
|
func (store *Store) AppendEvent(ctx context.Context, issueUUID domain.UUID, input EventInput) (Event, error) {
|
|
if !validUUID(issueUUID) || input.Kind == 0 || input.Compression < 1 || input.Compression > 2 || input.CreatedAt.IsZero() {
|
|
return Event{}, errors.New("invalid client event")
|
|
}
|
|
tx, err := store.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return Event{}, err
|
|
}
|
|
defer tx.Rollback()
|
|
var nextOrdinal, outputCharged, totalCharged, closeout uint64
|
|
err = tx.QueryRowContext(ctx, `SELECT next_local_ordinal, output_charged_bytes, total_charged_bytes, closeout_remaining_bytes FROM commands WHERE issue_uuid = ?`, issueUUID[:]).Scan(&nextOrdinal, &outputCharged, &totalCharged, &closeout)
|
|
if err == sql.ErrNoRows {
|
|
return Event{}, ErrUnknownCommand
|
|
}
|
|
if err != nil {
|
|
return Event{}, err
|
|
}
|
|
charge, err := EstimateCharge(ChargeInput{EncodedBytes: uint64(len(input.Payload)), SQLiteRows: 1, IndexEntries: 2})
|
|
if err != nil {
|
|
return Event{}, err
|
|
}
|
|
clientTotal, err := clientTotalCharge(ctx, tx)
|
|
if err != nil {
|
|
return Event{}, err
|
|
}
|
|
decision, err := CheckReservation(store.quotaLimits, ReservationState{CommandOutputCharged: outputCharged, CommandTotalCharged: totalCharged, ClientTotalCharged: clientTotal, CloseoutRemaining: closeout}, ReservationRequest{ChargedBytes: charge, Output: input.Output, UseCloseout: input.UseCloseout})
|
|
if err != nil {
|
|
return Event{}, err
|
|
}
|
|
ordinal := nextOrdinal
|
|
digest := immutableDigest(input.Payload)
|
|
_, err = tx.ExecContext(ctx, `INSERT INTO events(issue_uuid, local_ordinal, event_kind, compression, raw_bytes, charged_bytes, output, payload, payload_sha256, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, issueUUID[:], ordinal, input.Kind, input.Compression, input.RawBytes, charge, boolInt(input.Output), input.Payload, digest[:], input.CreatedAt.UnixNano())
|
|
if err != nil {
|
|
return Event{}, err
|
|
}
|
|
if _, err = tx.ExecContext(ctx, `UPDATE commands SET next_local_ordinal = ?, output_charged_bytes = ?, total_charged_bytes = ?, closeout_remaining_bytes = ? WHERE issue_uuid = ?`, ordinal+1, decision.CommandOutputCharged, decision.CommandTotalCharged, decision.CloseoutRemaining, issueUUID[:]); err != nil {
|
|
return Event{}, err
|
|
}
|
|
if err := updateClientTotalCharge(ctx, tx, decision.ClientTotalCharged); err != nil {
|
|
return Event{}, err
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return Event{}, err
|
|
}
|
|
return Event{IssueUUID: issueUUID, LocalOrdinal: ordinal, Kind: input.Kind, Compression: input.Compression, RawBytes: input.RawBytes, Payload: append([]byte(nil), input.Payload...), CreatedAt: input.CreatedAt}, nil
|
|
}
|
|
|
|
// AssignSendWindow atomically gives previously-unassigned events contiguous
|
|
// per-command wire sequences. Once assigned, they are immutable until Ack.
|
|
func (store *Store) AssignSendWindow(ctx context.Context, issueUUID domain.UUID, maximumEvents, maximumStoredBytes uint64) ([]Event, error) {
|
|
if maximumEvents == 0 || maximumStoredBytes == 0 {
|
|
return nil, nil
|
|
}
|
|
tx, err := store.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer tx.Rollback()
|
|
var next uint64
|
|
err = tx.QueryRowContext(ctx, `SELECT next_event_seq FROM commands WHERE issue_uuid = ?`, issueUUID[:]).Scan(&next)
|
|
if err == sql.ErrNoRows {
|
|
return nil, ErrUnknownCommand
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
rows, err := tx.QueryContext(ctx, `SELECT local_ordinal, length(payload) FROM events WHERE issue_uuid = ? AND event_seq IS NULL ORDER BY local_ordinal LIMIT ?`, issueUUID[:], maximumEvents)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var ordinals []uint64
|
|
for rows.Next() {
|
|
var ordinal, storedBytes uint64
|
|
if err := rows.Scan(&ordinal, &storedBytes); err != nil {
|
|
_ = rows.Close()
|
|
return nil, err
|
|
}
|
|
if storedBytes > maximumStoredBytes {
|
|
break
|
|
}
|
|
maximumStoredBytes -= storedBytes
|
|
ordinals = append(ordinals, ordinal)
|
|
}
|
|
if err := rows.Close(); err != nil {
|
|
return nil, err
|
|
}
|
|
for _, ordinal := range ordinals {
|
|
if _, err := tx.ExecContext(ctx, `UPDATE events SET event_seq = ? WHERE issue_uuid = ? AND local_ordinal = ? AND event_seq IS NULL`, next, issueUUID[:], ordinal); err != nil {
|
|
return nil, err
|
|
}
|
|
next++
|
|
}
|
|
if len(ordinals) > 0 {
|
|
if _, err := tx.ExecContext(ctx, `UPDATE commands SET next_event_seq = ? WHERE issue_uuid = ?`, next, issueUUID[:]); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
events, err := eventsBySequence(ctx, tx, issueUUID, 0)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return nil, err
|
|
}
|
|
return events, nil
|
|
}
|
|
|
|
func (store *Store) PendingEvents(ctx context.Context, issueUUID domain.UUID) ([]Event, error) {
|
|
return eventsBySequence(ctx, store.db, issueUUID, 0)
|
|
}
|
|
|
|
// Ack applies a cumulative server acknowledgement. Acknowledging an event that
|
|
// was never assigned is a protocol/storage boundary violation, never a hint to
|
|
// skip data.
|
|
func (store *Store) Ack(ctx context.Context, issueUUID domain.UUID, eventSeq uint64) error {
|
|
tx, err := store.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
var lastAck, next, outputCharged, totalCharged uint64
|
|
err = tx.QueryRowContext(ctx, `SELECT last_server_ack, next_event_seq, output_charged_bytes, total_charged_bytes FROM commands WHERE issue_uuid = ?`, issueUUID[:]).Scan(&lastAck, &next, &outputCharged, &totalCharged)
|
|
if err == sql.ErrNoRows {
|
|
return ErrUnknownCommand
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if eventSeq <= lastAck {
|
|
return tx.Commit()
|
|
}
|
|
if eventSeq >= next {
|
|
return ErrInvalidEventAck
|
|
}
|
|
var released, releasedOutput uint64
|
|
if err := tx.QueryRowContext(ctx, `SELECT COALESCE(sum(charged_bytes), 0), COALESCE(sum(CASE WHEN output = 1 THEN charged_bytes ELSE 0 END), 0) FROM events WHERE issue_uuid = ? AND event_seq <= ?`, issueUUID[:], eventSeq).Scan(&released, &releasedOutput); err != nil {
|
|
return err
|
|
}
|
|
if released > totalCharged || releasedOutput > outputCharged {
|
|
return fmt.Errorf("client spool charge counter mismatch")
|
|
}
|
|
clientTotal, err := clientTotalCharge(ctx, tx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if released > clientTotal {
|
|
return fmt.Errorf("client spool aggregate charge counter mismatch")
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `DELETE FROM events WHERE issue_uuid = ? AND event_seq <= ?`, issueUUID[:], eventSeq); err != nil {
|
|
return err
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `UPDATE commands SET last_server_ack = ?, output_charged_bytes = ?, total_charged_bytes = ? WHERE issue_uuid = ?`, eventSeq, outputCharged-releasedOutput, totalCharged-released, issueUUID[:]); err != nil {
|
|
return err
|
|
}
|
|
if err := updateClientTotalCharge(ctx, tx, clientTotal-released); err != nil {
|
|
return err
|
|
}
|
|
return tx.Commit()
|
|
}
|
|
|
|
func (store *Store) MarkTerminal(ctx context.Context, issueUUID domain.UUID, phase uint32) error {
|
|
if phase < 5 || phase > 11 {
|
|
return errors.New("invalid terminal command phase")
|
|
}
|
|
result, err := store.db.ExecContext(ctx, `UPDATE commands SET phase = ?, terminal = 1 WHERE issue_uuid = ?`, phase, issueUUID[:])
|
|
if err != nil {
|
|
return err
|
|
}
|
|
count, err := result.RowsAffected()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if count == 0 {
|
|
return ErrUnknownCommand
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// CleanupTerminal moves a fully acknowledged terminal command into the compact
|
|
// tombstone ledger and removes all command-owned spool data in one transaction.
|
|
func (store *Store) CleanupTerminal(ctx context.Context, issueUUID domain.UUID, acknowledgedAt time.Time) error {
|
|
if acknowledgedAt.IsZero() {
|
|
return errors.New("terminal acknowledgement time is required")
|
|
}
|
|
tx, err := store.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
var hash []byte
|
|
var terminal int
|
|
var totalCharged uint64
|
|
err = tx.QueryRowContext(ctx, `SELECT immutable_sha256, terminal, total_charged_bytes FROM commands WHERE issue_uuid = ?`, issueUUID[:]).Scan(&hash, &terminal, &totalCharged)
|
|
if err == sql.ErrNoRows {
|
|
return ErrUnknownCommand
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if terminal == 0 {
|
|
return errors.New("command is not terminal")
|
|
}
|
|
var pending uint64
|
|
if err := tx.QueryRowContext(ctx, `SELECT count(*) FROM events WHERE issue_uuid = ?`, issueUUID[:]).Scan(&pending); err != nil {
|
|
return err
|
|
}
|
|
if pending != 0 {
|
|
return ErrEventsPending
|
|
}
|
|
clientTotal, err := clientTotalCharge(ctx, tx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if totalCharged > clientTotal {
|
|
return fmt.Errorf("client spool aggregate charge counter mismatch")
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `INSERT INTO command_tombstones(issue_uuid, immutable_sha256, acknowledged_at) VALUES (?, ?, ?)`, issueUUID[:], hash, acknowledgedAt.UnixNano()); err != nil {
|
|
return err
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `DELETE FROM commands WHERE issue_uuid = ?`, issueUUID[:]); err != nil {
|
|
return err
|
|
}
|
|
if err := updateClientTotalCharge(ctx, tx, clientTotal-totalCharged); err != nil {
|
|
return err
|
|
}
|
|
trim, err := tombstonesToTrim(ctx, tx, store.tombstoneLimit)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `DELETE FROM command_tombstones WHERE issue_uuid IN (SELECT issue_uuid FROM command_tombstones ORDER BY acknowledged_at, issue_uuid LIMIT ?)`, trim); err != nil {
|
|
return err
|
|
}
|
|
return tx.Commit()
|
|
}
|
|
|
|
type queryer interface {
|
|
QueryContext(context.Context, string, ...any) (*sql.Rows, error)
|
|
}
|
|
|
|
func eventsBySequence(ctx context.Context, query queryer, issueUUID domain.UUID, after uint64) ([]Event, error) {
|
|
rows, err := query.QueryContext(ctx, `SELECT local_ordinal, event_seq, event_kind, compression, raw_bytes, payload, created_at FROM events WHERE issue_uuid = ? AND event_seq IS NOT NULL AND event_seq > ? ORDER BY event_seq`, issueUUID[:], after)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var events []Event
|
|
for rows.Next() {
|
|
var event Event
|
|
var created int64
|
|
event.IssueUUID = issueUUID
|
|
if err := rows.Scan(&event.LocalOrdinal, &event.EventSeq, &event.Kind, &event.Compression, &event.RawBytes, &event.Payload, &created); err != nil {
|
|
return nil, err
|
|
}
|
|
event.CreatedAt = time.Unix(0, created).UTC()
|
|
events = append(events, event)
|
|
}
|
|
return events, rows.Err()
|
|
}
|
|
|
|
func commandByUUID(ctx context.Context, query interface {
|
|
QueryRowContext(context.Context, string, ...any) *sql.Row
|
|
}, issueUUID domain.UUID) (Command, bool, error) {
|
|
var command Command
|
|
var hash []byte
|
|
var terminal int
|
|
err := query.QueryRowContext(ctx, `SELECT immutable_sha256, command_revision, phase, terminal FROM commands WHERE issue_uuid = ?`, issueUUID[:]).Scan(&hash, &command.Revision, &command.Phase, &terminal)
|
|
if err == sql.ErrNoRows {
|
|
return Command{}, false, nil
|
|
}
|
|
if err != nil {
|
|
return Command{}, false, err
|
|
}
|
|
if len(hash) != len(command.ImmutableSHA256) {
|
|
return Command{}, false, fmt.Errorf("invalid stored immutable command hash")
|
|
}
|
|
copy(command.ImmutableSHA256[:], hash)
|
|
command.IssueUUID = issueUUID
|
|
command.Terminal = terminal != 0
|
|
return command, true, nil
|
|
}
|
|
|
|
func tombstonesToTrim(ctx context.Context, tx *sql.Tx, limit uint64) (uint64, error) {
|
|
var count uint64
|
|
if err := tx.QueryRowContext(ctx, `SELECT count(*) FROM command_tombstones`).Scan(&count); err != nil {
|
|
return 0, err
|
|
}
|
|
if count <= limit {
|
|
return 0, nil
|
|
}
|
|
return count - limit, nil
|
|
}
|
|
|
|
func boolInt(value bool) int {
|
|
if value {
|
|
return 1
|
|
}
|
|
return 0
|
|
}
|
|
|
|
func clientTotalCharge(ctx context.Context, tx *sql.Tx) (uint64, error) {
|
|
var total uint64
|
|
if err := tx.QueryRowContext(ctx, `SELECT client_total_charged_bytes FROM spool_counters WHERE singleton = 1`).Scan(&total); err != nil {
|
|
return 0, err
|
|
}
|
|
return total, nil
|
|
}
|
|
|
|
func updateClientTotalCharge(ctx context.Context, tx *sql.Tx, total uint64) error {
|
|
_, err := tx.ExecContext(ctx, `UPDATE spool_counters SET client_total_charged_bytes = ? WHERE singleton = 1`, total)
|
|
return err
|
|
}
|
|
|
|
func validUUID(value domain.UUID) bool {
|
|
_, err := domain.ParseUUIDv7(value.String())
|
|
return err == nil
|
|
}
|
|
|
|
func isTerminalPhase(phase uint32) bool { return phase >= 5 && phase <= 11 }
|