Persist execution audit fields and enforce the in-flight guard in SQL
Findings 4 and 7 of REVIEW-2026-07-30.md, as a single user_version step (11 -> 12) because they alter the same table. confirmation_id was written by the INSERT and absent from all three executions SELECTs, so it always read back empty: "which confirmation authorized this destructive act" was unanswerable from the API. causation_id was set on the struct by the engine with no column to land in, and capability_version was missing entirely despite spec §4.1 and the precedent on confirmations. All three are now persisted and selected back. The one-in-flight-per-(capability, target) rule was check-then-insert with no constraint between, so two concurrent requests could both proceed. It is now a partial UNIQUE index; the engine's pre-insert query is demoted to an advisory fast path and a constraint violation maps to ErrExecutionInFlight (409), kept distinct from the idempotency-key replay case. Note the migration also rewrites pre-existing duplicate in-flight rows to status='unknown', keeping the lowest id per pair — creating the index would otherwise fail outright on any database holding stale started rows, which the old non-unique index permitted indefinitely. That is a write to existing audit rows, not just DDL. Rows predating this migration get capability_version=0, which is indistinguishable from a genuine 0; backfill is not possible. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Uea55zaiWuEByEDC4UBSdd
This commit is contained in:
kami
@@ -0,0 +1,153 @@
|
||||
package execution_test
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/kami/hexis/internal/domain"
|
||||
)
|
||||
|
||||
// The audit trail is only useful if it survives a round trip through storage:
|
||||
// "which confirmation authorized this destructive act" must be answerable from
|
||||
// a plain read of the execution, not just from the in-memory struct the engine
|
||||
// happened to return.
|
||||
func TestExecute_AuditFieldsRoundTripThroughStorage(t *testing.T) {
|
||||
prov := &fakeProvider{name: "fake"}
|
||||
eng, store := newTestEngine(t, prov)
|
||||
cap := mustCreateCapability(t, store, func(c *domain.Capability) {
|
||||
c.RequiresConfirmation = true
|
||||
c.Risk = "destructive"
|
||||
})
|
||||
|
||||
args := map[string]any{"foo": "bar"}
|
||||
conf, err := eng.CreateConfirmation(cap.ID, "ent_x", "kami", args)
|
||||
if err != nil {
|
||||
t.Fatalf("create confirmation: %v", err)
|
||||
}
|
||||
|
||||
res, err := eng.Execute(&domain.ExecuteRequest{
|
||||
CapabilityID: cap.ID,
|
||||
TargetEntityID: "ent_x",
|
||||
Arguments: args,
|
||||
ConfirmationID: conf.ID,
|
||||
CorrelationID: "corr_1",
|
||||
CausationID: "caus_1",
|
||||
IdempotencyKey: "idem_1",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("execute: %v", err)
|
||||
}
|
||||
|
||||
// This is what GET /api/v1/executions/{id} serves.
|
||||
got, err := store.GetExecution(res.Execution.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("get execution: %v", err)
|
||||
}
|
||||
if got.ConfirmationID != conf.ID {
|
||||
t.Errorf("confirmation_id: got %q, want %q", got.ConfirmationID, conf.ID)
|
||||
}
|
||||
if got.CausationID != "caus_1" {
|
||||
t.Errorf("causation_id: got %q, want %q", got.CausationID, "caus_1")
|
||||
}
|
||||
if got.CorrelationID != "corr_1" {
|
||||
t.Errorf("correlation_id: got %q, want %q", got.CorrelationID, "corr_1")
|
||||
}
|
||||
if got.CapabilityVersion != cap.Version {
|
||||
t.Errorf("capability_version: got %d, want %d", got.CapabilityVersion, cap.Version)
|
||||
}
|
||||
|
||||
// The idempotency-key read path serves the same record and must agree.
|
||||
byKey, err := store.GetExecutionByIdempotencyKey("idem_1")
|
||||
if err != nil {
|
||||
t.Fatalf("get by idempotency key: %v", err)
|
||||
}
|
||||
if byKey.ConfirmationID != conf.ID || byKey.CausationID != "caus_1" || byKey.CapabilityVersion != cap.Version {
|
||||
t.Errorf("idempotency read path lost audit fields: %+v", byKey)
|
||||
}
|
||||
}
|
||||
|
||||
// The in-flight guard used to be check-then-insert. Two concurrent identical
|
||||
// executes must produce exactly one execution row, with the loser reported as
|
||||
// already-in-flight rather than a 500.
|
||||
func TestExecute_ConcurrentDuplicateProducesExactlyOneRow(t *testing.T) {
|
||||
prov := &fakeProvider{name: "fake", delay: 150 * time.Millisecond}
|
||||
eng, store := newTestEngine(t, prov)
|
||||
cap := mustCreateCapability(t, store, nil)
|
||||
|
||||
const n = 4
|
||||
var wg sync.WaitGroup
|
||||
errs := make([]error, n)
|
||||
start := make(chan struct{})
|
||||
for i := 0; i < n; i++ {
|
||||
wg.Add(1)
|
||||
go func(i int) {
|
||||
defer wg.Done()
|
||||
<-start
|
||||
_, errs[i] = eng.Execute(&domain.ExecuteRequest{
|
||||
CapabilityID: cap.ID,
|
||||
TargetEntityID: "ent_x",
|
||||
})
|
||||
}(i)
|
||||
}
|
||||
close(start)
|
||||
wg.Wait()
|
||||
|
||||
succeeded := 0
|
||||
for i, err := range errs {
|
||||
switch {
|
||||
case err == nil:
|
||||
succeeded++
|
||||
case errors.Is(err, domain.ErrExecutionInFlight):
|
||||
default:
|
||||
t.Errorf("goroutine %d: unexpected error %v (want nil or ErrExecutionInFlight)", i, err)
|
||||
}
|
||||
}
|
||||
if succeeded != 1 {
|
||||
t.Errorf("expected exactly 1 successful execute, got %d", succeeded)
|
||||
}
|
||||
|
||||
var rows int
|
||||
if err := store.DB().QueryRow(
|
||||
`SELECT COUNT(*) FROM executions WHERE capability_id = ? AND target_entity_id = ?`,
|
||||
cap.ID, "ent_x",
|
||||
).Scan(&rows); err != nil {
|
||||
t.Fatalf("count executions: %v", err)
|
||||
}
|
||||
if rows != 1 {
|
||||
t.Errorf("expected exactly 1 execution row, got %d", rows)
|
||||
}
|
||||
}
|
||||
|
||||
// Directly exercise the partial unique index, bypassing the engine's advisory
|
||||
// pre-check, so the constraint itself is what is under test.
|
||||
func TestStore_SecondInFlightRowRejectedByIndex(t *testing.T) {
|
||||
_, store := newTestEngine(t, &fakeProvider{name: "fake"})
|
||||
now := time.Now().UTC()
|
||||
|
||||
mk := func(status domain.ExecutionStatus) *domain.Execution {
|
||||
return &domain.Execution{
|
||||
ID: domain.NewExecutionID(),
|
||||
CapabilityID: "cap_1",
|
||||
TargetEntityID: "ent_x",
|
||||
Status: status,
|
||||
CreatedAt: now,
|
||||
UpdatedAt: now,
|
||||
}
|
||||
}
|
||||
|
||||
if err := store.CreateExecution(mk(domain.ExecutionStarted)); err != nil {
|
||||
t.Fatalf("first insert: %v", err)
|
||||
}
|
||||
if err := store.CreateExecution(mk(domain.ExecutionStarted)); !errors.Is(err, domain.ErrExecutionInFlight) {
|
||||
t.Fatalf("second in-flight insert: got %v, want ErrExecutionInFlight", err)
|
||||
}
|
||||
// The index is partial: terminal rows for the same pair stay insertable.
|
||||
if err := store.CreateExecution(mk(domain.ExecutionSucceeded)); err != nil {
|
||||
t.Fatalf("terminal-status insert should be allowed: %v", err)
|
||||
}
|
||||
if err := store.CreateExecution(mk(domain.ExecutionSucceeded)); err != nil {
|
||||
t.Fatalf("second terminal-status insert should be allowed: %v", err)
|
||||
}
|
||||
}
|
||||
@@ -53,7 +53,10 @@ func (e *Engine) Execute(req *domain.ExecuteRequest) (*ExecuteResult, error) {
|
||||
}
|
||||
}
|
||||
|
||||
// One in-flight execution per (capability_id, target_entity_id).
|
||||
// One in-flight execution per (capability_id, target_entity_id). This check
|
||||
// is advisory — it gives a clean error before doing any work — but the
|
||||
// authoritative guard is the partial unique index enforced at INSERT below,
|
||||
// which closes the check-then-insert window between the two.
|
||||
if _, err := e.store.GetInFlightExecution(req.CapabilityID, req.TargetEntityID); err == nil {
|
||||
return nil, domain.ErrExecutionInFlight
|
||||
}
|
||||
@@ -67,6 +70,7 @@ func (e *Engine) Execute(req *domain.ExecuteRequest) (*ExecuteResult, error) {
|
||||
exec := &domain.Execution{
|
||||
ID: domain.NewExecutionID(),
|
||||
CapabilityID: req.CapabilityID,
|
||||
CapabilityVersion: capability.Version,
|
||||
TargetEntityID: req.TargetEntityID,
|
||||
EntityVersion: req.EntityVersion,
|
||||
Arguments: req.Arguments,
|
||||
|
||||
+45
-12
@@ -141,6 +141,27 @@ var migrations = []string{
|
||||
state TEXT NOT NULL DEFAULT 'pending'
|
||||
)`,
|
||||
`CREATE INDEX IF NOT EXISTS idx_executions_inflight ON executions(capability_id, target_entity_id, status)`,
|
||||
// Audit columns missing from executions (spec §4.1) plus a partial UNIQUE
|
||||
// index enforcing one in-flight execution per (capability, target).
|
||||
// Pre-existing duplicate in-flight rows would make the index creation fail,
|
||||
// so they are resolved to 'unknown' first — that is the spec's outcome for
|
||||
// an execution whose side effect can no longer be determined.
|
||||
`ALTER TABLE executions ADD COLUMN causation_id TEXT NOT NULL DEFAULT '';
|
||||
ALTER TABLE executions ADD COLUMN capability_version INTEGER NOT NULL DEFAULT 0;
|
||||
UPDATE executions SET status = 'unknown'
|
||||
WHERE status = 'started'
|
||||
AND id NOT IN (
|
||||
SELECT MIN(id) FROM executions WHERE status = 'started'
|
||||
GROUP BY capability_id, target_entity_id
|
||||
);
|
||||
DROP INDEX IF EXISTS idx_executions_inflight;
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS idx_executions_inflight
|
||||
ON executions(capability_id, target_entity_id) WHERE status = 'started';`,
|
||||
// Supports the entity_id filter on GET /api/v1/executions (spec §4.5). The
|
||||
// `since` cursor is the row's implicit rowid, which cannot appear in an
|
||||
// index (SQLite indexes it as the btree key already), so ordering and
|
||||
// range-scanning by rowid needs no index of its own.
|
||||
`CREATE INDEX IF NOT EXISTS idx_executions_entity ON executions(target_entity_id)`,
|
||||
}
|
||||
|
||||
const timeFmt = "2006-01-02T15:04:05.999999999Z07:00"
|
||||
@@ -292,12 +313,20 @@ func (s *Store) CreateExecution(e *domain.Execution) error {
|
||||
evidence, _ := json.Marshal(e.ResolutionEvidence)
|
||||
|
||||
_, err := s.db.Exec(
|
||||
`INSERT INTO executions (id, capability_id, target_entity_id, entity_version, arguments, requested_by, origin, idempotency_key, confirmation_id, status, result, error, resolution_evidence, correlation_id, created_at, updated_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
e.ID, e.CapabilityID, e.TargetEntityID, e.EntityVersion, string(args), string(reqBy), string(origin), nullString(e.IdempotencyKey), e.ConfirmationID, string(e.Status), string(result), e.Error, string(evidence), e.CorrelationID, formatTime(e.CreatedAt), formatTime(e.UpdatedAt),
|
||||
`INSERT INTO executions (id, capability_id, capability_version, target_entity_id, entity_version, arguments, requested_by, origin, idempotency_key, confirmation_id, status, result, error, resolution_evidence, correlation_id, causation_id, created_at, updated_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
e.ID, e.CapabilityID, e.CapabilityVersion, e.TargetEntityID, e.EntityVersion, string(args), string(reqBy), string(origin), nullString(e.IdempotencyKey), e.ConfirmationID, string(e.Status), string(result), e.Error, string(evidence), e.CorrelationID, e.CausationID, formatTime(e.CreatedAt), formatTime(e.UpdatedAt),
|
||||
)
|
||||
if err != nil {
|
||||
if strings.Contains(err.Error(), "UNIQUE") {
|
||||
// The partial unique index on (capability_id, target_entity_id)
|
||||
// WHERE status='started' is the in-flight guard, not a replay.
|
||||
if strings.Contains(err.Error(), "idempotency_key") {
|
||||
return domain.ErrIdempotencyReplay
|
||||
}
|
||||
if strings.Contains(err.Error(), "capability_id") || strings.Contains(err.Error(), "target_entity_id") {
|
||||
return domain.ErrExecutionInFlight
|
||||
}
|
||||
return domain.ErrIdempotencyReplay
|
||||
}
|
||||
return err
|
||||
@@ -310,12 +339,12 @@ func (s *Store) GetExecution(id string) (*domain.Execution, error) {
|
||||
defer s.mu.RUnlock()
|
||||
|
||||
row := s.db.QueryRow(
|
||||
`SELECT id, capability_id, target_entity_id, entity_version, arguments, requested_by, origin, COALESCE(idempotency_key,''), status, result, COALESCE(error,''), resolution_evidence, COALESCE(correlation_id,''), created_at, updated_at
|
||||
`SELECT id, capability_id, capability_version, target_entity_id, entity_version, arguments, requested_by, origin, COALESCE(idempotency_key,''), COALESCE(confirmation_id,''), status, result, COALESCE(error,''), resolution_evidence, COALESCE(correlation_id,''), COALESCE(causation_id,''), created_at, updated_at
|
||||
FROM executions WHERE id = ?`, id,
|
||||
)
|
||||
e := &domain.Execution{}
|
||||
var args, reqBy, origin, idempKey, status, result, errStr, evidence, corrID, createdAt, updatedAt string
|
||||
err := row.Scan(&e.ID, &e.CapabilityID, &e.TargetEntityID, &e.EntityVersion, &args, &reqBy, &origin, &idempKey, &status, &result, &errStr, &evidence, &corrID, &createdAt, &updatedAt)
|
||||
var args, reqBy, origin, idempKey, confID, status, result, errStr, evidence, corrID, causID, createdAt, updatedAt string
|
||||
err := row.Scan(&e.ID, &e.CapabilityID, &e.CapabilityVersion, &e.TargetEntityID, &e.EntityVersion, &args, &reqBy, &origin, &idempKey, &confID, &status, &result, &errStr, &evidence, &corrID, &causID, &createdAt, &updatedAt)
|
||||
if err == sql.ErrNoRows {
|
||||
return nil, domain.ErrExecutionNotFound
|
||||
}
|
||||
@@ -328,9 +357,11 @@ func (s *Store) GetExecution(id string) (*domain.Execution, error) {
|
||||
json.Unmarshal([]byte(result), &e.Result)
|
||||
json.Unmarshal([]byte(evidence), &e.ResolutionEvidence)
|
||||
e.IdempotencyKey = idempKey
|
||||
e.ConfirmationID = confID
|
||||
e.Status = domain.ExecutionStatus(status)
|
||||
e.Error = errStr
|
||||
e.CorrelationID = corrID
|
||||
e.CausationID = causID
|
||||
e.CreatedAt = parseTime(createdAt)
|
||||
e.UpdatedAt = parseTime(updatedAt)
|
||||
if e.Arguments == nil {
|
||||
@@ -366,12 +397,12 @@ func (s *Store) GetExecutionByIdempotencyKey(key string) (*domain.Execution, err
|
||||
defer s.mu.RUnlock()
|
||||
|
||||
row := s.db.QueryRow(
|
||||
`SELECT id, capability_id, target_entity_id, entity_version, arguments, requested_by, origin, idempotency_key, status, result, error, resolution_evidence, correlation_id, created_at, updated_at
|
||||
`SELECT id, capability_id, capability_version, target_entity_id, entity_version, arguments, requested_by, origin, COALESCE(idempotency_key,''), COALESCE(confirmation_id,''), status, result, COALESCE(error,''), resolution_evidence, COALESCE(correlation_id,''), COALESCE(causation_id,''), created_at, updated_at
|
||||
FROM executions WHERE idempotency_key = ?`, key,
|
||||
)
|
||||
e := &domain.Execution{}
|
||||
var args, reqBy, origin, idempKey, status, result, errStr, evidence, corrID, createdAt, updatedAt string
|
||||
err := row.Scan(&e.ID, &e.CapabilityID, &e.TargetEntityID, &e.EntityVersion, &args, &reqBy, &origin, &idempKey, &status, &result, &errStr, &evidence, &corrID, &createdAt, &updatedAt)
|
||||
var args, reqBy, origin, idempKey, confID, status, result, errStr, evidence, corrID, causID, createdAt, updatedAt string
|
||||
err := row.Scan(&e.ID, &e.CapabilityID, &e.CapabilityVersion, &e.TargetEntityID, &e.EntityVersion, &args, &reqBy, &origin, &idempKey, &confID, &status, &result, &errStr, &evidence, &corrID, &causID, &createdAt, &updatedAt)
|
||||
if err == sql.ErrNoRows {
|
||||
return nil, domain.ErrExecutionNotFound
|
||||
}
|
||||
@@ -384,9 +415,11 @@ func (s *Store) GetExecutionByIdempotencyKey(key string) (*domain.Execution, err
|
||||
json.Unmarshal([]byte(result), &e.Result)
|
||||
json.Unmarshal([]byte(evidence), &e.ResolutionEvidence)
|
||||
e.IdempotencyKey = idempKey
|
||||
e.ConfirmationID = confID
|
||||
e.Status = domain.ExecutionStatus(status)
|
||||
e.Error = errStr
|
||||
e.CorrelationID = corrID
|
||||
e.CausationID = causID
|
||||
e.CreatedAt = parseTime(createdAt)
|
||||
e.UpdatedAt = parseTime(updatedAt)
|
||||
return e, nil
|
||||
@@ -415,12 +448,12 @@ func (s *Store) GetInFlightExecution(capabilityID, targetEntityID string) (*doma
|
||||
// already hold it (read or write).
|
||||
func (s *Store) getExecutionLocked(id string) (*domain.Execution, error) {
|
||||
row := s.db.QueryRow(
|
||||
`SELECT id, capability_id, target_entity_id, entity_version, arguments, requested_by, origin, COALESCE(idempotency_key,''), status, result, COALESCE(error,''), resolution_evidence, COALESCE(correlation_id,''), created_at, updated_at
|
||||
`SELECT id, capability_id, capability_version, target_entity_id, entity_version, arguments, requested_by, origin, COALESCE(idempotency_key,''), COALESCE(confirmation_id,''), status, result, COALESCE(error,''), resolution_evidence, COALESCE(correlation_id,''), COALESCE(causation_id,''), created_at, updated_at
|
||||
FROM executions WHERE id = ?`, id,
|
||||
)
|
||||
e := &domain.Execution{}
|
||||
var args, reqBy, origin, idempKey, status, result, errStr, evidence, corrID, createdAt, updatedAt string
|
||||
err := row.Scan(&e.ID, &e.CapabilityID, &e.TargetEntityID, &e.EntityVersion, &args, &reqBy, &origin, &idempKey, &status, &result, &errStr, &evidence, &corrID, &createdAt, &updatedAt)
|
||||
var args, reqBy, origin, idempKey, confID, status, result, errStr, evidence, corrID, causID, createdAt, updatedAt string
|
||||
err := row.Scan(&e.ID, &e.CapabilityID, &e.CapabilityVersion, &e.TargetEntityID, &e.EntityVersion, &args, &reqBy, &origin, &idempKey, &confID, &status, &result, &errStr, &evidence, &corrID, &causID, &createdAt, &updatedAt)
|
||||
if err == sql.ErrNoRows {
|
||||
return nil, domain.ErrExecutionNotFound
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user