feat: apply terminal reconciliation cleanup
This commit is contained in:
@@ -345,6 +345,83 @@ func (store *Store) CleanupTerminal(ctx context.Context, issueUUID domain.UUID,
|
||||
return tx.Commit()
|
||||
}
|
||||
|
||||
// DiscardTerminal removes a server-confirmed terminal command even when its
|
||||
// local event rows have not been acknowledged. The compact tombstone is kept so
|
||||
// a late duplicate dispatch cannot execute it again. It is intentionally only
|
||||
// for a ReconcileResult that authoritatively tells the client to discard local
|
||||
// terminal state; ordinary terminal cleanup must use CleanupTerminal.
|
||||
func (store *Store) DiscardTerminal(ctx context.Context, issueUUID domain.UUID, acknowledgedAt time.Time) error {
|
||||
if !validUUID(issueUUID) || acknowledgedAt.IsZero() {
|
||||
return errors.New("invalid terminal discard")
|
||||
}
|
||||
tx, err := store.db.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer tx.Rollback()
|
||||
|
||||
var hash []byte
|
||||
var revision uint64
|
||||
var phase uint32
|
||||
var terminal int
|
||||
var totalCharged uint64
|
||||
err = tx.QueryRowContext(ctx, `SELECT immutable_sha256, command_revision, phase, terminal, total_charged_bytes FROM commands WHERE issue_uuid = ?`, issueUUID[:]).Scan(&hash, &revision, &phase, &terminal, &totalCharged)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
var tombstoneHash []byte
|
||||
err = tx.QueryRowContext(ctx, `SELECT immutable_sha256 FROM command_tombstones WHERE issue_uuid = ?`, issueUUID[:]).Scan(&tombstoneHash)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return ErrUnknownCommand
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if len(tombstoneHash) != 32 {
|
||||
return fmt.Errorf("invalid terminal tombstone")
|
||||
}
|
||||
return tx.Commit()
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if terminal == 0 {
|
||||
return errors.New("command is not terminal")
|
||||
}
|
||||
clientTotal, err := clientTotalCharge(ctx, tx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if totalCharged > clientTotal {
|
||||
return fmt.Errorf("client spool aggregate charge counter mismatch")
|
||||
}
|
||||
var existingHash []byte
|
||||
err = tx.QueryRowContext(ctx, `SELECT immutable_sha256 FROM command_tombstones WHERE issue_uuid = ?`, issueUUID[:]).Scan(&existingHash)
|
||||
if err == nil && (len(existingHash) != len(hash) || string(existingHash) != string(hash)) {
|
||||
return ErrCommandConflict
|
||||
}
|
||||
if err != nil && !errors.Is(err, sql.ErrNoRows) {
|
||||
return err
|
||||
}
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
if _, err := tx.ExecContext(ctx, `INSERT INTO command_tombstones(issue_uuid, immutable_sha256, command_revision, terminal_lifecycle, acknowledged_at) VALUES (?, ?, ?, ?, ?)`, issueUUID[:], hash, revision, phase, 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)
|
||||
}
|
||||
|
||||
@@ -154,6 +154,40 @@ func TestTerminalCleanupTombstonesAndConflicts_BH_CLIENT_02(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestDiscardServerConfirmedTerminalDropsPendingEvents_HP_CLIENT_10(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
store := openTestStore(t, ctx, filepath.Join(t.TempDir(), "spool"), DefaultTombstoneLimit)
|
||||
issue := testUUID(t, "019c46f1-1d02-7000-8000-000000000013")
|
||||
command := testCommand(issue, []byte("server-confirmed terminal"))
|
||||
now := time.Date(2026, time.September, 6, 12, 0, 0, 0, time.UTC)
|
||||
if _, err := store.AcceptCommand(ctx, command, now); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := store.AppendEvent(ctx, issue, EventInput{Kind: 1, Compression: 1, Payload: []byte("unacknowledged local terminal output"), CreatedAt: now}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := store.MarkTerminal(ctx, issue, 5); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := store.DiscardTerminal(ctx, issue, now); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if pending, err := store.PendingEvents(ctx, issue); err != nil || len(pending) != 0 {
|
||||
t.Fatalf("discarded command pending events = %#v, %v", pending, err)
|
||||
}
|
||||
snapshot, err := store.ReconcileSnapshot(ctx)
|
||||
if err != nil || len(snapshot.GetRetainedCommands()) != 1 || !snapshot.GetRetainedCommands()[0].GetTombstoned() {
|
||||
t.Fatalf("discarded command snapshot = %#v, %v", snapshot, err)
|
||||
}
|
||||
if err := store.DiscardTerminal(ctx, issue, now.Add(time.Second)); err != nil {
|
||||
t.Fatalf("repeated discard = %v", err)
|
||||
}
|
||||
if _, err := store.AcceptCommand(ctx, command, now); !errors.Is(err, ErrAlreadyExecuted) {
|
||||
t.Fatalf("duplicate after discard error = %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func openTestStore(t *testing.T, ctx context.Context, directory string, tombstoneLimit uint64) *Store {
|
||||
t.Helper()
|
||||
store, err := Open(ctx, Options{DataDir: directory, BusyTimeout: time.Second, TombstoneLimit: tombstoneLimit})
|
||||
|
||||
Reference in New Issue
Block a user