weichen
adcd8fefd2
Why: Account disable could be followed by late Backend business writes, API key creation could implicitly create users, account bootstrap locks could invert, and directory cursors were process-local or leaked out-of-scope changes. Evidence: Local go test ./... and isolated PostgreSQL package suites passed on the staged candidate; exact committed-head SELF replay and stage 6 human review remain pending. Impact: Adds final account transactions and external-effect checks across HTTP, Host, Bus and run manager; tightens API key issuance and persists versioned cursor signing. Delegated RDM rename still needs the separately proposed real-operator contract, so A1 and stage 6 are not complete. Refs: Documents U3 r1, approved A1-A4 amendment fb3a821, pending delegated-projection amendment 5783940, Dev feature/iam-u3-account-directory-recovery.
330 lines
13 KiB
Go
330 lines
13 KiB
Go
package store
|
||
|
||
import (
|
||
"database/sql"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"time"
|
||
)
|
||
|
||
// 协作存储 U5（KTD4/KTD5）：会话执行租约 + 持久命令收件箱。
|
||
//
|
||
// 租约：conversation_executions 一行一会话。TryClaimExecution 空闲/租约到期
|
||
// 才可占用；接管（含同实例重新占用）递增 epoch——旧 owner 的续租与结算按
|
||
//（instance, epoch）双重校验被拒（fencing）。lease_until 用应用侧 UTC ISO
|
||
// （单库写入方一致即可；跨库时钟偏移属部署合同，见 runbook）。
|
||
//
|
||
// 命令：run_commands 收件箱承载跨实例取消/HITL（KTD5）。command_id 幂等；
|
||
// pending→consumed 由消费方原子认领（PG FOR UPDATE SKIP LOCKED；SQLite 单写
|
||
// 者条件更新）。命令不删除——审计与重放判定保留。
|
||
|
||
// ErrExecutionBusy 会话执行权被其他实例持有且租约未到期。
|
||
var ErrExecutionBusy = errors.New("store: execution owned by another instance")
|
||
|
||
// ErrFenced 旧执行代次的写入被围栏拒绝（epoch 已被接管递增）。
|
||
var ErrFenced = errors.New("store: fenced by newer execution epoch")
|
||
|
||
// ExecutionState 当前执行租约快照（诊断/测试）。
|
||
type ExecutionState struct {
|
||
ConversationID string
|
||
InstanceID string
|
||
RunID string
|
||
Epoch int64
|
||
LeaseUntil string
|
||
}
|
||
|
||
// TryClaimExecution 占用（或租约到期后接管）会话执行权。
|
||
// 返回接管后的 epoch（接管必然 +1）。被他人持有且未到期 → ErrExecutionBusy。
|
||
func (s *Store) TryClaimExecution(conversationID, instanceID, runID string, leaseSec int) (int64, error) {
|
||
tx, err := s.db.Begin()
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
defer tx.Rollback()
|
||
if err := s.lockAdmission(tx); err != nil { // 复用全局账本锁串行化清点
|
||
return 0, err
|
||
}
|
||
epoch, err := s.claimExecutionTx(tx, conversationID, instanceID, runID, leaseSec)
|
||
if err != nil {
|
||
return epoch, err
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return 0, err
|
||
}
|
||
return epoch, nil
|
||
}
|
||
|
||
// TryClaimExecutionAccountChecked locks the actor before the admission ledger
|
||
// and verifies the conversation/run owner in the same final write transaction.
|
||
func (s *Store) TryClaimExecutionAccountChecked(actor, conversationID, instanceID, runID string, leaseSec int, allowImplicitDefault bool) (int64, error) {
|
||
var epoch int64
|
||
err := s.withEffectiveAccountWrite(actor, allowImplicitDefault, func(tx *sql.Tx, _ Account) error {
|
||
if err := s.lockAdmission(tx); err != nil {
|
||
return err
|
||
}
|
||
var owner string
|
||
if err := tx.QueryRow(s.q(`SELECT owner FROM conversations WHERE id=?`), conversationID).Scan(&owner); errors.Is(err, sql.ErrNoRows) {
|
||
return ErrNotFound
|
||
} else if err != nil {
|
||
return err
|
||
}
|
||
if owner != actor {
|
||
return ErrNotFound
|
||
}
|
||
if runID != "" {
|
||
var one int
|
||
if err := tx.QueryRow(s.q(`SELECT 1 FROM runs WHERE id=? AND owner=? AND conversation_id=?`), runID, actor, conversationID).Scan(&one); errors.Is(err, sql.ErrNoRows) {
|
||
return ErrNotFound
|
||
} else if err != nil {
|
||
return err
|
||
}
|
||
}
|
||
var err error
|
||
epoch, err = s.claimExecutionTx(tx, conversationID, instanceID, runID, leaseSec)
|
||
return err
|
||
})
|
||
return epoch, err
|
||
}
|
||
|
||
func (s *Store) claimExecutionTx(tx *sql.Tx, conversationID, instanceID, runID string, leaseSec int) (int64, error) {
|
||
if leaseSec <= 0 {
|
||
leaseSec = 90
|
||
}
|
||
now := time.Now().UTC()
|
||
until := now.Add(time.Duration(leaseSec) * time.Second).UTC().Format(time.RFC3339Nano)
|
||
var inst, lease string
|
||
var epoch int64
|
||
err := tx.QueryRow(s.q(`SELECT instance_id, epoch, lease_until FROM conversation_executions WHERE conversation_id=?`), conversationID).Scan(&inst, &epoch, &lease)
|
||
switch {
|
||
case errors.Is(err, sql.ErrNoRows):
|
||
if _, err := tx.Exec(s.q(`INSERT INTO conversation_executions (conversation_id, instance_id, run_id, epoch, lease_until, updated_at) VALUES (?,?,?,1,?,?)`),
|
||
conversationID, instanceID, runID, until, nowISO()); err != nil {
|
||
return 0, err
|
||
}
|
||
return 1, nil
|
||
case err != nil:
|
||
return 0, err
|
||
}
|
||
if inst != instanceID && lease > nowISO() {
|
||
return epoch, fmt.Errorf("conv %s: %w", conversationID, ErrExecutionBusy)
|
||
}
|
||
// 接管（到期或同实例重占）：epoch+1，fencing 由此产生。
|
||
if _, err := tx.Exec(s.q(`UPDATE conversation_executions SET instance_id=?, run_id=?, epoch=epoch+1, lease_until=?, updated_at=? WHERE conversation_id=?`),
|
||
instanceID, runID, until, nowISO(), conversationID); err != nil {
|
||
return 0, err
|
||
}
|
||
return epoch + 1, nil
|
||
}
|
||
|
||
// RenewExecution 续租（fencing：instance+epoch 双校验）。假 = 已被接管或释放。
|
||
func (s *Store) RenewExecution(conversationID, instanceID string, epoch int64, leaseSec int) bool {
|
||
if leaseSec <= 0 {
|
||
leaseSec = 90
|
||
}
|
||
until := time.Now().UTC().Add(time.Duration(leaseSec) * time.Second).UTC().Format(time.RFC3339Nano)
|
||
res, err := s.db.Exec(s.q(`UPDATE conversation_executions SET lease_until=?, updated_at=? WHERE conversation_id=? AND instance_id=? AND epoch=?`),
|
||
until, nowISO(), conversationID, instanceID, epoch)
|
||
if err != nil {
|
||
return false
|
||
}
|
||
n, _ := res.RowsAffected()
|
||
return n > 0
|
||
}
|
||
|
||
// RenewExecutionAccountChecked only extends the lease while the conversation
|
||
// owner is effective in the same transaction as the conditional UPDATE.
|
||
func (s *Store) RenewExecutionAccountChecked(actor, conversationID, instanceID string, epoch int64, leaseSec int, allowImplicitDefault bool) bool {
|
||
if leaseSec <= 0 {
|
||
leaseSec = 90
|
||
}
|
||
until := time.Now().UTC().Add(time.Duration(leaseSec) * time.Second).UTC().Format(time.RFC3339Nano)
|
||
updated := false
|
||
err := s.withEffectiveAccountWrite(actor, allowImplicitDefault, func(tx *sql.Tx, _ Account) error {
|
||
var owner string
|
||
if err := tx.QueryRow(s.q(`SELECT owner FROM conversations WHERE id=?`), conversationID).Scan(&owner); errors.Is(err, sql.ErrNoRows) {
|
||
return ErrNotFound
|
||
} else if err != nil {
|
||
return err
|
||
}
|
||
if owner != actor {
|
||
return ErrNotFound
|
||
}
|
||
res, err := tx.Exec(s.q(`UPDATE conversation_executions SET lease_until=?, updated_at=? WHERE conversation_id=? AND instance_id=? AND epoch=?`),
|
||
until, nowISO(), conversationID, instanceID, epoch)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
n, err := res.RowsAffected()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
updated = n == 1
|
||
return nil
|
||
})
|
||
return err == nil && updated
|
||
}
|
||
|
||
// ReleaseExecution 主动释放（优雅停机）：行保留、租约置为已过期——epoch 保留，
|
||
// 下一次占用仍递增（fencing 单调）。
|
||
func (s *Store) ReleaseExecution(conversationID, instanceID string, epoch int64) {
|
||
_, _ = s.db.Exec(s.q(`UPDATE conversation_executions SET lease_until=?, updated_at=? WHERE conversation_id=? AND instance_id=? AND epoch=?`),
|
||
time.Now().UTC().Add(-time.Second).UTC().Format(time.RFC3339Nano), nowISO(), conversationID, instanceID, epoch)
|
||
}
|
||
|
||
// GetExecution 读取当前租约（无行 = 从未占用）。
|
||
func (s *Store) GetExecution(conversationID string) (ExecutionState, error) {
|
||
var e ExecutionState
|
||
err := s.db.QueryRow(s.q(`SELECT conversation_id, instance_id, COALESCE(run_id,''), epoch, lease_until FROM conversation_executions WHERE conversation_id=?`), conversationID).
|
||
Scan(&e.ConversationID, &e.InstanceID, &e.RunID, &e.Epoch, &e.LeaseUntil)
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return e, ErrNotFound
|
||
}
|
||
return e, err
|
||
}
|
||
|
||
// ExecutionFenceOK 校验 (instance, epoch) 仍是当前执行权（结算/事件写入的
|
||
// 前置检查；事务内校验见 SettleRunFenced）。
|
||
func (s *Store) ExecutionFenceOK(conversationID, instanceID string, epoch int64) bool {
|
||
var n int
|
||
if err := s.db.QueryRow(s.q(`SELECT COUNT(*) FROM conversation_executions WHERE conversation_id=? AND instance_id=? AND epoch=?`), conversationID, instanceID, epoch).Scan(&n); err != nil {
|
||
return false
|
||
}
|
||
return n > 0
|
||
}
|
||
|
||
// SettleRunFenced 带执行围栏的终态结算（KTD4：旧 owner 恢复后不能提交）。
|
||
// fenced=true 表示 (instance,epoch) 已不是当前执行权——run 未被改动，调用方
|
||
// 应将本地 run 视为被接管（不写终态、事件标注接管）。无租约行（SQLite 本地
|
||
// 档/未启用执行权）恒放行。
|
||
|
||
// RunCommand 收件箱命令。
|
||
type RunCommand struct {
|
||
CommandID string
|
||
Kind string // cancel | hitl
|
||
RunID string
|
||
ConversationID string
|
||
Epoch int64
|
||
Payload string
|
||
CreatedAt string
|
||
}
|
||
|
||
// EnqueueCommandInTx 与业务提交同事务投递命令（KTD5/U5 余项：outbox——
|
||
// 批准数据与待投递命令原子提交）。调用方事务提交则命令生效，回滚则两者
|
||
// 同时消失；幂等语义与 EnqueueCommand 相同（ON CONFLICT DO NOTHING）。
|
||
func (s *Store) EnqueueCommandInTx(tx *sql.Tx, commandID, kind, runID, conversationID string, epoch int64, payload any) error {
|
||
raw, err := json.Marshal(payload)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
_, err = tx.Exec(s.q(`INSERT INTO run_commands (command_id, kind, run_id, conversation_id, epoch, payload, state, created_at)
|
||
VALUES (?,?,?,?,?,?,'pending',?)
|
||
ON CONFLICT(command_id) DO NOTHING`),
|
||
commandID, kind, runID, conversationID, epoch, string(raw), nowISO())
|
||
if err != nil {
|
||
return fmt.Errorf("enqueue command %s (tx): %w", commandID, err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// EnqueueCommand 投递跨实例命令（幂等：同 command_id 重复投递不产生新行）。
|
||
func (s *Store) EnqueueCommand(commandID, kind, runID, conversationID string, epoch int64, payload any) error {
|
||
raw, err := json.Marshal(payload)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
_, err = s.db.Exec(s.q(`INSERT INTO run_commands (command_id, kind, run_id, conversation_id, epoch, payload, state, created_at)
|
||
VALUES (?,?,?,?,?,?,'pending',?)
|
||
ON CONFLICT(command_id) DO NOTHING`),
|
||
commandID, kind, runID, conversationID, epoch, string(raw), nowISO())
|
||
if err != nil {
|
||
return fmt.Errorf("enqueue command %s: %w", commandID, err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// EnqueueCommandAccountChecked resolves the run owner and conversation from
|
||
// persisted rows in the final account-guarded transaction. A client-supplied
|
||
// run ID can locate a command target but cannot authorize another owner's run.
|
||
func (s *Store) EnqueueCommandAccountChecked(actor, commandID, kind, runID string, epoch int64, payload any, allowImplicitDefault bool) error {
|
||
return s.withEffectiveAccountWrite(actor, allowImplicitDefault, func(tx *sql.Tx, _ Account) error {
|
||
var owner, conversationID string
|
||
query := s.q(`SELECT owner,conversation_id FROM runs WHERE id=?`)
|
||
if s.Dialect() == "postgres" {
|
||
query += ` FOR UPDATE`
|
||
}
|
||
if err := tx.QueryRow(query, runID).Scan(&owner, &conversationID); errors.Is(err, sql.ErrNoRows) {
|
||
return ErrNotFound
|
||
} else if err != nil {
|
||
return err
|
||
}
|
||
if owner != actor {
|
||
return ErrNotFound
|
||
}
|
||
return s.EnqueueCommandInTx(tx, commandID, kind, runID, conversationID, epoch, payload)
|
||
})
|
||
}
|
||
|
||
// ClaimPendingCommands 认领待处理命令（原子 pending→consumed）。
|
||
// PG：FOR UPDATE SKIP LOCKED（多实例并发认领不重复）；SQLite：单写者条件更新。
|
||
// 只认领「本实例相关」的命令由调用方过滤——收件箱按执行权路由：命令在
|
||
// 被执行实例消费（cancel/hitl 目标 run 的当前 owner）。
|
||
func (s *Store) ClaimPendingCommands(consumer string, limit int) ([]RunCommand, error) {
|
||
if limit <= 0 {
|
||
limit = 16
|
||
}
|
||
tx, err := s.db.Begin()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer tx.Rollback()
|
||
var rows *sql.Rows
|
||
if s.dialect.name() == "postgres" {
|
||
rows, err = tx.Query(`SELECT command_id, kind, run_id, conversation_id, epoch, payload, created_at
|
||
FROM run_commands WHERE state='pending' ORDER BY created_at
|
||
LIMIT $1 FOR UPDATE SKIP LOCKED`, limit)
|
||
} else {
|
||
rows, err = tx.Query(s.q(`SELECT command_id, kind, run_id, conversation_id, epoch, payload, created_at
|
||
FROM run_commands WHERE state='pending' ORDER BY created_at LIMIT ?`), limit)
|
||
}
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
var out []RunCommand
|
||
for rows.Next() {
|
||
var c RunCommand
|
||
if err := rows.Scan(&c.CommandID, &c.Kind, &c.RunID, &c.ConversationID, &c.Epoch, &c.Payload, &c.CreatedAt); err != nil {
|
||
rows.Close()
|
||
return nil, err
|
||
}
|
||
out = append(out, c)
|
||
}
|
||
rows.Close()
|
||
if err := rows.Err(); err != nil {
|
||
return nil, err
|
||
}
|
||
for _, c := range out {
|
||
res, err := tx.Exec(s.q(`UPDATE run_commands SET state='consumed', consumed_at=?, consumer=? WHERE command_id=? AND state='pending'`),
|
||
nowISO(), consumer, c.CommandID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if n, _ := res.RowsAffected(); n == 0 {
|
||
return nil, fmt.Errorf("command %s claimed concurrently", c.CommandID)
|
||
}
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return nil, err
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// CommandState 查询单条命令状态（测试/诊断）。
|
||
func (s *Store) CommandState(commandID string) (state, consumer string, err error) {
|
||
err = s.db.QueryRow(s.q(`SELECT state, COALESCE(consumer,'') FROM run_commands WHERE command_id=?`), commandID).Scan(&state, &consumer)
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return "", "", ErrNotFound
|
||
}
|
||
return
|
||
}
|