306 lines
12 KiB
Go
306 lines
12 KiB
Go
package store
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"errors"
|
|
"fmt"
|
|
"sort"
|
|
"time"
|
|
|
|
rvboxv1 "github.com/rvbox/rvbox/gen/go/rvbox/v1"
|
|
"github.com/rvbox/rvbox/internal/domain"
|
|
)
|
|
|
|
// ReconcileTargets returns the server's complete non-terminal view for the
|
|
// client. It is a read-only snapshot used to tell a reconnecting agent which
|
|
// UUIDs and durable cursors must be compared before fresh dispatch is enabled.
|
|
func (store *Store) ReconcileTargets(ctx context.Context, clientID string) ([]*rvboxv1.ReconcileTarget, error) {
|
|
if clientID == "" {
|
|
return nil, errors.New("client ID is required")
|
|
}
|
|
rows, err := store.db.QueryContext(ctx, `SELECT issue_uuid, last_event_seq, revision, immutable_request_sha256
|
|
FROM commands WHERE client_id = ? AND lifecycle BETWEEN 1 AND 4 ORDER BY issue_time, issue_uuid`, clientID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var result []*rvboxv1.ReconcileTarget
|
|
for rows.Next() {
|
|
var issue, digest []byte
|
|
var eventSeq, revision uint64
|
|
if err := rows.Scan(&issue, &eventSeq, &revision, &digest); err != nil {
|
|
return nil, err
|
|
}
|
|
if len(issue) != 16 || len(digest) != 32 || revision == 0 {
|
|
return nil, ErrInvalidSegmentRecord
|
|
}
|
|
var parsed domain.UUID
|
|
copy(parsed[:], issue)
|
|
if _, err := domain.ParseUUIDv7(parsed.String()); err != nil {
|
|
return nil, ErrInvalidSegmentRecord
|
|
}
|
|
result = append(result, &rvboxv1.ReconcileTarget{IssueUuid: parsed.String(), LastServerEventSeq: eventSeq, CommandRevision: revision, ImmutableRequestSha256: append([]byte(nil), digest...)})
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
// ReconcileClientSnapshot compares client evidence with this client's durable
|
|
// server rows. It intentionally performs no lifecycle mutation yet: callers
|
|
// receive only the safe local terminate/discard instructions.
|
|
func (store *Store) ReconcileClientSnapshot(ctx context.Context, clientID string, snapshot *rvboxv1.ReconcileSnapshot) (*rvboxv1.ReconcileResult, error) {
|
|
return store.reconcileClientSnapshot(ctx, clientID, 0, snapshot)
|
|
}
|
|
|
|
// ReconcileClientSnapshotForSession applies the complete bidirectional
|
|
// reconciliation matrix while the registering session is fenced. Queued or
|
|
// dispatched rows absent from the client's complete snapshot are safely
|
|
// requeued; accepted/running rows absent from the snapshot are interrupted and
|
|
// incidented. Retained non-terminal rows are retargeted to this generation so
|
|
// late events from the previous connection cannot advance the command.
|
|
func (store *Store) ReconcileClientSnapshotForSession(ctx context.Context, clientID string, generation uint64, snapshot *rvboxv1.ReconcileSnapshot) (*rvboxv1.ReconcileResult, error) {
|
|
if generation == 0 {
|
|
return nil, errors.New("session generation is required")
|
|
}
|
|
return store.reconcileClientSnapshot(ctx, clientID, generation, snapshot)
|
|
}
|
|
|
|
type reconcileServerRow struct {
|
|
issue domain.UUID
|
|
lifecycle rvboxv1.CommandLifecycle
|
|
revision uint64
|
|
lastSeq uint64
|
|
hash []byte
|
|
target sql.NullInt64
|
|
}
|
|
|
|
func (store *Store) reconcileClientSnapshot(ctx context.Context, clientID string, generation uint64, snapshot *rvboxv1.ReconcileSnapshot) (*rvboxv1.ReconcileResult, error) {
|
|
if clientID == "" {
|
|
return nil, errors.New("client ID is required")
|
|
}
|
|
if snapshot == nil {
|
|
return nil, errors.New("missing client reconciliation snapshot")
|
|
}
|
|
clientRows := make(map[string]*rvboxv1.ReconcileCommandState, len(snapshot.GetRetainedCommands()))
|
|
for _, client := range snapshot.GetRetainedCommands() {
|
|
if client == nil || client.GetCommandRevision() == 0 || len(client.GetImmutableRequestSha256()) != 32 {
|
|
return nil, fmt.Errorf("invalid client reconciliation row")
|
|
}
|
|
issue, err := domain.ParseUUIDv7(client.GetIssueUuid())
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
key := issue.String()
|
|
if _, exists := clientRows[key]; exists {
|
|
return nil, fmt.Errorf("duplicate client reconciliation row %s", key)
|
|
}
|
|
clientRows[key] = client
|
|
}
|
|
|
|
store.writeMu.Lock()
|
|
database, err := store.openDatabase()
|
|
if err != nil {
|
|
store.writeMu.Unlock()
|
|
return nil, err
|
|
}
|
|
tx, err := database.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
store.writeMu.Unlock()
|
|
return nil, err
|
|
}
|
|
defer tx.Rollback()
|
|
serverRows, tombstones, err := loadReconcileRows(ctx, tx, clientID)
|
|
if err != nil {
|
|
store.writeMu.Unlock()
|
|
return nil, err
|
|
}
|
|
result := &rvboxv1.ReconcileResult{}
|
|
var incidents []reconcileIncident
|
|
for key, client := range clientRows {
|
|
row, present := serverRows[key]
|
|
input := domain.ReconcileInput{ClientEvidence: domain.ClientEvidenceRetained, ClientLifecycle: client.GetLifecycle(), ClientRevision: domain.CommandRevision(client.GetCommandRevision()), ClientLastEventSeq: client.GetLastClientEventSeq(), ImmutableHashMatches: false}
|
|
if client.GetTombstoned() {
|
|
input.ClientEvidence = domain.ClientEvidenceTombstone
|
|
}
|
|
if present {
|
|
input.ServerPresent = true
|
|
input.ServerLifecycle = row.lifecycle
|
|
input.ServerRevision = domain.CommandRevision(row.revision)
|
|
input.ServerLastEventSeq = row.lastSeq
|
|
input.ImmutableHashMatches = len(row.hash) == 32 && bytesEqual(row.hash, client.GetImmutableRequestSha256())
|
|
} else if tombstoneHash, found := tombstones[key]; found {
|
|
input.ServerHasTombstone = true
|
|
input.ImmutableHashMatches = len(tombstoneHash) == 32 && bytesEqual(tombstoneHash, client.GetImmutableRequestSha256())
|
|
} else {
|
|
// Once the server has neither a live row nor a tombstone, there is
|
|
// no immutable request hash left to compare. The client evidence is
|
|
// still useful: a retained non-terminal row must be terminated and
|
|
// a retained terminal row may be discarded. Treat the evidence as
|
|
// structurally valid here; hash equality is required whenever the
|
|
// server still has a row or tombstone to compare against.
|
|
input.ImmutableHashMatches = true
|
|
}
|
|
decision, decisionErr := domain.DecideReconciliation(input)
|
|
if decisionErr != nil {
|
|
incidents = append(incidents, reconcileIncident{issue: parseIssueUnchecked(key), summary: decisionErr.Error(), dataLoss: false})
|
|
continue
|
|
}
|
|
switch decision.Action {
|
|
case domain.ReconcileTerminateLocal:
|
|
result.TerminateLocalIssueUuids = append(result.TerminateLocalIssueUuids, key)
|
|
if decision.RecordIncident {
|
|
incidents = append(incidents, reconcileIncident{issue: parseIssueUnchecked(key), summary: "client and server command state disagree; local process must be terminated", dataLoss: true})
|
|
}
|
|
case domain.ReconcileDiscardLocalTerminal:
|
|
result.DiscardLocalTerminalIssueUuids = append(result.DiscardLocalTerminalIssueUuids, key)
|
|
case domain.ReconcileResumeDelivery:
|
|
if generation != 0 && row.target.Valid && uint64(row.target.Int64) != generation || generation != 0 && !row.target.Valid {
|
|
if _, err := tx.ExecContext(ctx, `UPDATE commands SET target_session_generation = ? WHERE issue_uuid = ? AND client_id = ? AND lifecycle BETWEEN 2 AND 4`, generation, row.issue[:], clientID); err != nil {
|
|
store.writeMu.Unlock()
|
|
return nil, err
|
|
}
|
|
}
|
|
case domain.ReconcileInterruptServerSuppressReplay:
|
|
if err := interruptReconcileRow(ctx, tx, row, clientID, "client tombstone contradicts non-terminal server command"); err != nil {
|
|
store.writeMu.Unlock()
|
|
return nil, err
|
|
}
|
|
incidents = append(incidents, reconcileIncident{issue: row.issue, summary: "client tombstone contradicts non-terminal server command", dataLoss: true})
|
|
}
|
|
}
|
|
for key, row := range serverRows {
|
|
if _, present := clientRows[key]; present {
|
|
continue
|
|
}
|
|
decision, decisionErr := domain.DecideReconciliation(domain.ReconcileInput{ServerPresent: true, ServerLifecycle: row.lifecycle, ServerRevision: domain.CommandRevision(row.revision), ServerLastEventSeq: row.lastSeq, ClientEvidence: domain.ClientEvidenceAbsent})
|
|
if decisionErr != nil {
|
|
incidents = append(incidents, reconcileIncident{issue: row.issue, summary: decisionErr.Error(), dataLoss: false})
|
|
continue
|
|
}
|
|
switch decision.Action {
|
|
case domain.ReconcileRequeue:
|
|
if _, err := tx.ExecContext(ctx, `UPDATE commands SET lifecycle = 1, target_session_generation = NULL WHERE issue_uuid = ? AND client_id = ? AND lifecycle IN (1, 2)`, row.issue[:], clientID); err != nil {
|
|
store.writeMu.Unlock()
|
|
return nil, err
|
|
}
|
|
case domain.ReconcileInterruptServerClientStateLoss:
|
|
if err := interruptReconcileRow(ctx, tx, row, clientID, "client lost accepted/running command state"); err != nil {
|
|
store.writeMu.Unlock()
|
|
return nil, err
|
|
}
|
|
incidents = append(incidents, reconcileIncident{issue: row.issue, summary: "client lost accepted/running command state", dataLoss: true})
|
|
}
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
store.writeMu.Unlock()
|
|
return nil, err
|
|
}
|
|
store.writeMu.Unlock()
|
|
for _, incident := range incidents {
|
|
if err := store.recordReconcileIncident(ctx, clientID, incident); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
sort.Strings(result.TerminateLocalIssueUuids)
|
|
sort.Strings(result.DiscardLocalTerminalIssueUuids)
|
|
return result, nil
|
|
}
|
|
|
|
type reconcileIncident struct {
|
|
issue domain.UUID
|
|
summary string
|
|
dataLoss bool
|
|
}
|
|
|
|
func loadReconcileRows(ctx context.Context, tx *sql.Tx, clientID string) (map[string]reconcileServerRow, map[string][]byte, error) {
|
|
rows, err := tx.QueryContext(ctx, `SELECT issue_uuid, lifecycle, revision, last_event_seq, immutable_request_sha256, target_session_generation FROM commands WHERE client_id = ?`, clientID)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
commands := make(map[string]reconcileServerRow)
|
|
for rows.Next() {
|
|
var encoded, hash []byte
|
|
var lifecycle uint32
|
|
var row reconcileServerRow
|
|
if err := rows.Scan(&encoded, &lifecycle, &row.revision, &row.lastSeq, &hash, &row.target); err != nil {
|
|
_ = rows.Close()
|
|
return nil, nil, err
|
|
}
|
|
if len(encoded) != 16 || len(hash) != 32 || lifecycle == 0 || lifecycle > 11 || row.revision == 0 {
|
|
_ = rows.Close()
|
|
return nil, nil, ErrInvalidSegmentRecord
|
|
}
|
|
copy(row.issue[:], encoded)
|
|
if _, err := domain.ParseUUIDv7(row.issue.String()); err != nil {
|
|
return nil, nil, ErrInvalidSegmentRecord
|
|
}
|
|
row.lifecycle = rvboxv1.CommandLifecycle(lifecycle)
|
|
row.hash = append([]byte(nil), hash...)
|
|
commands[row.issue.String()] = row
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
_ = rows.Close()
|
|
return nil, nil, err
|
|
}
|
|
if err := rows.Close(); err != nil {
|
|
return nil, nil, err
|
|
}
|
|
tombstoneRows, err := tx.QueryContext(ctx, `SELECT issue_uuid, immutable_sha256 FROM command_tombstones WHERE client_id = ?`, clientID)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
defer tombstoneRows.Close()
|
|
tombstones := make(map[string][]byte)
|
|
for tombstoneRows.Next() {
|
|
var encoded, hash []byte
|
|
if err := tombstoneRows.Scan(&encoded, &hash); err != nil {
|
|
return nil, nil, err
|
|
}
|
|
var issue domain.UUID
|
|
if len(encoded) != 16 || len(hash) != 32 {
|
|
return nil, nil, ErrInvalidSegmentRecord
|
|
}
|
|
copy(issue[:], encoded)
|
|
tombstones[issue.String()] = append([]byte(nil), hash...)
|
|
}
|
|
return commands, tombstones, tombstoneRows.Err()
|
|
}
|
|
|
|
func interruptReconcileRow(ctx context.Context, tx *sql.Tx, row reconcileServerRow, clientID, detail string) error {
|
|
_, err := tx.ExecContext(ctx, `UPDATE commands SET lifecycle = 9, terminal_time = ?, revision = revision + 1, target_session_generation = NULL WHERE issue_uuid = ? AND client_id = ? AND lifecycle BETWEEN 2 AND 4`, time.Now().UTC().UnixNano(), row.issue[:], clientID)
|
|
return err
|
|
}
|
|
|
|
func parseIssueUnchecked(value string) domain.UUID {
|
|
issue, _ := domain.ParseUUIDv7(value)
|
|
return issue
|
|
}
|
|
|
|
func bytesEqual(left, right []byte) bool {
|
|
if len(left) != len(right) {
|
|
return false
|
|
}
|
|
for index := range left {
|
|
if left[index] != right[index] {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
func (store *Store) recordReconcileIncident(ctx context.Context, clientID string, incident reconcileIncident) error {
|
|
if incident.issue == (domain.UUID{}) {
|
|
return errors.New("invalid reconciliation incident issue")
|
|
}
|
|
id, err := domain.NewUUIDv7()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
issue := [16]byte(incident.issue)
|
|
_, err = store.RecordIncident(ctx, IncidentInput{IncidentUUID: [16]byte(id), DetectedAt: time.Now().UTC(), Kind: IncidentCounterMismatch, Scope: IncidentScopeCommand, ScopeKey: incident.issue.String(), ClientID: clientID, IssueUUID: &issue, Summary: incident.summary, Evidence: []byte(incident.summary), DataLoss: incident.dataLoss, AutomaticallyRepairable: false})
|
|
return err
|
|
}
|