cabbage/rvbox
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: deliver durable live signal intents

Browse Source
This commit is contained in:
cabbage
2026-09-06 10:45:20 +00:00
parent ad564de23f
commit 26ac4c1cac
8 changed files with 227 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/server/store/event.go
+26 -1
View File
@@ -245,7 +245,32 @@ retained_compressed_bytes = retained_compressed_bytes + ?, output_charged_bytes
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())
var acknowledged sql.Result
acknowledged, 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 affected int64
affected, err = acknowledged.RowsAffected()
if err == nil && affected == 0 {
err = ErrInvalidSegmentRecord
}
}
}
}
if err == nil && event.EventType == 8 {
var wire rvboxv1.CommandEvent
if unmarshalErr := proto.Unmarshal(event.Payload, &wire); unmarshalErr != nil || wire.GetSignalResult() == nil || wire.GetSignalResult().GetCommandRevision() == 0 || wire.GetSignalResult().GetSignal() < rvboxv1.SignalKind_SIGNAL_HUP || wire.GetSignalResult().GetSignal() > rvboxv1.SignalKind_SIGNAL_USR2 {
err = ErrInvalidSegmentRecord
} else {
var acknowledged sql.Result
result := wire.GetSignalResult()
acknowledged, err = tx.ExecContext(ctx, `UPDATE signal_intents SET acknowledged = 1 WHERE issue_uuid = ? AND command_revision = ? AND signal = ?`, event.IssueUUID[:], result.GetCommandRevision(), result.GetSignal())
if err == nil {
var affected int64
affected, err = acknowledged.RowsAffected()
if err == nil && affected == 0 {
err = ErrInvalidSegmentRecord
}
}
}
}
if err == nil {
internal/server/store/migrations.go
+16 -1
View File
@@ -13,7 +13,10 @@ type migration struct {
sql string
}
var migrations = []migration{{version: 1, sql: schemaV1}}
var migrations = []migration{
{version: 1, sql: schemaV1},
{version: 2, sql: schemaV2},
}
func applyMigrations(ctx context.Context, db *sql.DB) error {
if _, err := db.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS schema_migrations (
@@ -181,3 +184,15 @@ CREATE TABLE storage_counters (
) STRICT;
INSERT INTO storage_counters(singleton, command_charged_bytes, charge_version) VALUES (1, 0, 1);
`
const schemaV2 = `
CREATE TABLE signal_intents (
request_uuid BLOB PRIMARY KEY CHECK(length(request_uuid) = 16),
issue_uuid BLOB NOT NULL REFERENCES commands(issue_uuid) ON DELETE CASCADE CHECK(length(issue_uuid) = 16),
signal INTEGER NOT NULL CHECK(signal BETWEEN 1 AND 6),
command_revision INTEGER NOT NULL CHECK(command_revision > 0),
acknowledged INTEGER NOT NULL DEFAULT 0 CHECK(acknowledged IN (0,1)),
created_at INTEGER NOT NULL
) STRICT;
CREATE INDEX signal_intents_pending ON signal_intents(issue_uuid, command_revision, signal, acknowledged, created_at, request_uuid);
`
internal/server/store/signal.go
+92 -6
View File
@@ -27,12 +27,19 @@ type SignalResult struct {
CommandRevision uint64
Duplicate bool
Cancelled bool
Pending bool
}
// SignalCommand applies the only locally-completable signal operation: a
// queued command can be cancelled before dispatch. For dispatched/accepted/
// running work the revisioned remote delivery path is intentionally not
// claimed until a live session control queue is available.
type SignalIntent struct {
IssueUUID domain.UUID
CommandRevision uint64
Signal rvboxv1.SignalKind
}
// SignalCommand cancels queued work locally or records a revision-bound,
// replayable signal intent for dispatched/accepted/running work. The session
// layer later exposes that intent and the client acknowledges it as a durable
// command event.
func (store *Store) SignalCommand(ctx context.Context, input SignalInput) (SignalResult, error) {
if input.IssueUUID == (domain.UUID{}) || input.ClientID == "" || input.RequestUUID == (domain.UUID{}) || input.Signal < rvboxv1.SignalKind_SIGNAL_HUP || input.Signal > rvboxv1.SignalKind_SIGNAL_USR2 || input.Hash == [32]byte{} || input.OccurredAt.IsZero() {
return SignalResult{}, errors.New("invalid signal request")
@@ -60,16 +67,63 @@ func (store *Store) SignalCommand(ctx context.Context, input SignalInput) (Signa
defer tx.Rollback()
var lifecycle uint32
var revision uint64
err = tx.QueryRowContext(ctx, `SELECT lifecycle, revision FROM commands WHERE issue_uuid = ? AND client_id = ?`, input.IssueUUID[:], input.ClientID).Scan(&lifecycle, &revision)
var targetGeneration sql.NullInt64
err = tx.QueryRowContext(ctx, `SELECT lifecycle, revision, target_session_generation FROM commands WHERE issue_uuid = ? AND client_id = ?`, input.IssueUUID[:], input.ClientID).Scan(&lifecycle, &revision, &targetGeneration)
if errors.Is(err, sql.ErrNoRows) {
return SignalResult{}, ErrCommandNotFound
}
if err != nil {
return SignalResult{}, err
}
if lifecycle != uint32(rvboxv1.CommandLifecycle_COMMAND_QUEUED) {
if lifecycle != uint32(rvboxv1.CommandLifecycle_COMMAND_QUEUED) && (lifecycle < uint32(rvboxv1.CommandLifecycle_COMMAND_DISPATCHED) || lifecycle > uint32(rvboxv1.CommandLifecycle_COMMAND_RUNNING) || !targetGeneration.Valid || targetGeneration.Int64 <= 0) {
return SignalResult{}, ErrSignalDeliveryUnavailable
}
if lifecycle != uint32(rvboxv1.CommandLifecycle_COMMAND_QUEUED) {
charge, chargeErr := EstimateCharge(ChargeInput{SQLiteRows: 1, IndexEntries: 1})
if chargeErr != nil {
return SignalResult{}, chargeErr
}
var commandCharged, closeout, clientCharged, serverCharged uint64
if err := tx.QueryRowContext(ctx, `SELECT commands.charged_bytes, commands.closeout_remaining_bytes, clients.charged_bytes, storage_counters.command_charged_bytes
FROM commands JOIN clients ON clients.client_id = commands.client_id JOIN storage_counters ON storage_counters.singleton = 1
WHERE commands.issue_uuid = ? AND commands.client_id = ?`, input.IssueUUID[:], input.ClientID).Scan(&commandCharged, &closeout, &clientCharged, &serverCharged); err != nil {
return SignalResult{}, err
}
freeBytes, freeErr := store.freeSpaceProbe.AvailableBytes(store.dataDir)
if freeErr != nil {
return SignalResult{}, freeErr
}
reservation, reserveErr := CheckReservation(store.quotaLimits, ReservationState{CommandTotalCharged: commandCharged, CloseoutRemaining: closeout, ClientTotalCharged: clientCharged, ServerTotalCharged: serverCharged, FilesystemFreeBytes: freeBytes}, ReservationRequest{ChargedBytes: charge})
if reserveErr != nil {
return SignalResult{}, reserveErr
}
if _, err := tx.ExecContext(ctx, `INSERT INTO signal_intents (request_uuid, issue_uuid, signal, command_revision, acknowledged, created_at) VALUES (?, ?, ?, ?, 0, ?)`, input.RequestUUID[:], input.IssueUUID[:], input.Signal, revision, input.OccurredAt.UTC().UnixNano()); err != nil {
return SignalResult{}, err
}
if _, err := tx.ExecContext(ctx, `UPDATE commands SET charged_bytes = ?, closeout_remaining_bytes = ? WHERE issue_uuid = ? AND charged_bytes = ?`, reservation.CommandTotalCharged, reservation.CloseoutRemaining, input.IssueUUID[:], commandCharged); err != nil {
return SignalResult{}, err
}
if _, err := tx.ExecContext(ctx, `UPDATE clients SET charged_bytes = ? WHERE client_id = ? AND charged_bytes = ?`, reservation.ClientTotalCharged, input.ClientID, clientCharged); err != nil {
return SignalResult{}, err
}
if _, err := tx.ExecContext(ctx, `UPDATE storage_counters SET command_charged_bytes = ? WHERE singleton = 1 AND command_charged_bytes = ?`, reservation.ServerTotalCharged, serverCharged); err != nil {
return SignalResult{}, err
}
result := make([]byte, 8)
binary.BigEndian.PutUint64(result, revision)
if _, err := tx.ExecContext(ctx, `INSERT INTO control_mutations (request_uuid, method, owner_kind, owner_id, target, immutable_sha256, assigned_revision, result, created_at) VALUES (?, ?, 'command', ?, ?, ?, ?, ?, ?)`, input.RequestUUID[:], method, input.IssueUUID.String(), target, input.Hash[:], revision, result, input.OccurredAt.UTC().UnixNano()); err != nil {
return SignalResult{}, err
}
auditPayload := []byte{byte(input.Signal)}
auditHash := sha256.Sum256(auditPayload)
if _, err := tx.ExecContext(ctx, `INSERT INTO audit_events (occurred_at, source, action, outcome, compression, payload, raw_bytes, stored_bytes, sha256) VALUES (?, 'control', ?, 'success', 1, ?, ?, ?, ?)`, input.OccurredAt.UTC().UnixNano(), method, auditPayload, len(auditPayload), len(auditPayload), auditHash[:]); err != nil {
return SignalResult{}, err
}
if err := tx.Commit(); err != nil {
return SignalResult{}, err
}
return SignalResult{CommandRevision: revision, Pending: true}, nil
}
if revision == ^uint64(0) {
return SignalResult{}, errors.New("command revision exhausted")
}
@@ -96,3 +150,35 @@ func (store *Store) SignalCommand(ctx context.Context, input SignalInput) (Signa
}
return SignalResult{CommandRevision: nextRevision, Cancelled: true}, nil
}
func (store *Store) PendingSignals(ctx context.Context, clientID string, generation uint64) ([]SignalIntent, error) {
if clientID == "" || generation == 0 {
return nil, errors.New("client ID and generation are required")
}
database, err := store.openDatabase()
if err != nil {
return nil, err
}
rows, err := database.QueryContext(ctx, `SELECT i.issue_uuid, i.command_revision, i.signal
FROM signal_intents i JOIN commands c ON c.issue_uuid = i.issue_uuid
WHERE c.client_id = ? AND c.target_session_generation = ? AND c.lifecycle BETWEEN 2 AND 4 AND i.acknowledged = 0
ORDER BY i.created_at, i.request_uuid`, clientID, generation)
if err != nil {
return nil, err
}
defer rows.Close()
var result []SignalIntent
for rows.Next() {
var encoded []byte
var intent SignalIntent
if err := rows.Scan(&encoded, &intent.CommandRevision, &intent.Signal); err != nil {
return nil, err
}
if len(encoded) != len(domain.UUID{}) || intent.CommandRevision == 0 || intent.Signal < rvboxv1.SignalKind_SIGNAL_HUP || intent.Signal > rvboxv1.SignalKind_SIGNAL_USR2 {
return nil, ErrInvalidSegmentRecord
}
copy(intent.IssueUUID[:], encoded)
result = append(result, intent)
}
return result, rows.Err()
}