You've already forked agentic-coding-workflow
feat(workspace,jobs): emit sync_aborted_on_restart audit + enrich worktree.pruned metadata
- workspace.ResetStuckSyncStatuses now returns affected IDs so app.New can record one workspace.sync_aborted_on_restart audit per reset row - worktree_prune audit metadata gains workspace_id and path alongside existing branch - updated fakeRepo + integration test for new ResetStuckSyncStatuses signature - add TestWorktreePrune_AuditMetadataContainsWorkspaceIDAndPath
This commit is contained in:
+15
-3
@@ -154,9 +154,21 @@ func New(ctx context.Context, cfg *config.Config, dist embed.FS) (*App, error) {
|
|||||||
})
|
})
|
||||||
|
|
||||||
wsRepo := workspace.NewPostgresRepository(pool)
|
wsRepo := workspace.NewPostgresRepository(pool)
|
||||||
// 进程启动时把卡死的 cloning/syncing 复位为 error,避免行永远卡住
|
// 进程启动时把卡死的 cloning/syncing 复位为 error,避免行永远卡住;
|
||||||
if err := wsRepo.ResetStuckSyncStatuses(ctx); err != nil {
|
// 每个被复位的工作区写一条 workspace.sync_aborted_on_restart audit。
|
||||||
|
if abortedIDs, err := wsRepo.ResetStuckSyncStatuses(ctx); err != nil {
|
||||||
log.Warn("reset stuck workspace sync statuses", "err", err.Error())
|
log.Warn("reset stuck workspace sync statuses", "err", err.Error())
|
||||||
|
} else {
|
||||||
|
for _, wsID := range abortedIDs {
|
||||||
|
if rerr := auditRec.Record(ctx, audit.Entry{
|
||||||
|
Action: "workspace.sync_aborted_on_restart",
|
||||||
|
TargetType: "workspace",
|
||||||
|
TargetID: wsID.String(),
|
||||||
|
}); rerr != nil {
|
||||||
|
log.Warn("workspace.sync_aborted_on_restart audit failed",
|
||||||
|
"workspace_id", wsID, "err", rerr.Error())
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pa := projectAccessAdapter{p: projectSvc}
|
pa := projectAccessAdapter{p: projectSvc}
|
||||||
@@ -614,7 +626,7 @@ func (a worktreePruneAdapter) enrichWorktrees(ctx context.Context, rows []*works
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (a worktreePruneAdapter) enrichOne(ctx context.Context, w *workspace.Worktree) runners.WorktreeRow {
|
func (a worktreePruneAdapter) enrichOne(ctx context.Context, w *workspace.Worktree) runners.WorktreeRow {
|
||||||
row := runners.WorktreeRow{ID: w.ID, Branch: w.Branch, Path: w.Path}
|
row := runners.WorktreeRow{ID: w.ID, WorkspaceID: w.WorkspaceID, Branch: w.Branch, Path: w.Path}
|
||||||
ws, err := a.wsRepo.GetWorkspaceByID(ctx, w.WorkspaceID)
|
ws, err := a.wsRepo.GetWorkspaceByID(ctx, w.WorkspaceID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
a.log.Warn("worktree_prune.adapter.lookup_workspace_failed",
|
a.log.Warn("worktree_prune.adapter.lookup_workspace_failed",
|
||||||
|
|||||||
@@ -345,7 +345,8 @@ func TestPlan3_WorkspaceClosedLoop(t *testing.T) {
|
|||||||
})
|
})
|
||||||
|
|
||||||
wsRepo := workspace.NewPostgresRepository(pool)
|
wsRepo := workspace.NewPostgresRepository(pool)
|
||||||
require.NoError(t, wsRepo.ResetStuckSyncStatuses(ctx))
|
_, err = wsRepo.ResetStuckSyncStatuses(ctx)
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
pa := projectAccessAdapterTest{p: projectSvc}
|
pa := projectAccessAdapterTest{p: projectSvc}
|
||||||
wsSvc := workspace.NewWorkspaceService(wsRepo, auditRec, enc, gitr, pa, nil, dataDir)
|
wsSvc := workspace.NewWorkspaceService(wsRepo, auditRec, enc, gitr, pa, nil, dataDir)
|
||||||
|
|||||||
@@ -15,11 +15,12 @@ import (
|
|||||||
// MainPath come from a JOIN performed in the production adapter (workspace
|
// MainPath come from a JOIN performed in the production adapter (workspace
|
||||||
// + project), not from the worktrees table directly.
|
// + project), not from the worktrees table directly.
|
||||||
type WorktreeRow struct {
|
type WorktreeRow struct {
|
||||||
ID uuid.UUID
|
ID uuid.UUID
|
||||||
OwnerID uuid.UUID
|
WorkspaceID uuid.UUID
|
||||||
Branch string
|
OwnerID uuid.UUID
|
||||||
MainPath string
|
Branch string
|
||||||
Path string
|
MainPath string
|
||||||
|
Path string
|
||||||
}
|
}
|
||||||
|
|
||||||
// WorktreePruneRepo is the minimal repository surface used by this runner.
|
// WorktreePruneRepo is the minimal repository surface used by this runner.
|
||||||
@@ -128,7 +129,9 @@ func (r *WorktreePrune) executionPhase(ctx context.Context) {
|
|||||||
TargetType: "worktree",
|
TargetType: "worktree",
|
||||||
TargetID: w.ID.String(),
|
TargetID: w.ID.String(),
|
||||||
Metadata: map[string]any{
|
Metadata: map[string]any{
|
||||||
"branch": w.Branch,
|
"workspace_id": w.WorkspaceID.String(),
|
||||||
|
"branch": w.Branch,
|
||||||
|
"path": w.Path,
|
||||||
},
|
},
|
||||||
}); rerr != nil {
|
}); rerr != nil {
|
||||||
r.log.Warn("worktree_prune.audit_failed", "worktree_id", w.ID, "err", rerr.Error())
|
r.log.Warn("worktree_prune.audit_failed", "worktree_id", w.ID, "err", rerr.Error())
|
||||||
|
|||||||
@@ -10,10 +10,20 @@ import (
|
|||||||
"github.com/stretchr/testify/assert"
|
"github.com/stretchr/testify/assert"
|
||||||
"github.com/stretchr/testify/require"
|
"github.com/stretchr/testify/require"
|
||||||
|
|
||||||
|
"github.com/yan1h/agent-coding-workflow/internal/audit"
|
||||||
"github.com/yan1h/agent-coding-workflow/internal/infra/notify"
|
"github.com/yan1h/agent-coding-workflow/internal/infra/notify"
|
||||||
"github.com/yan1h/agent-coding-workflow/internal/jobs/runners"
|
"github.com/yan1h/agent-coding-workflow/internal/jobs/runners"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
type recordingRecorder struct {
|
||||||
|
entries []audit.Entry
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *recordingRecorder) Record(_ context.Context, e audit.Entry) error {
|
||||||
|
r.entries = append(r.entries, e)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
// fakeWorktreeRepo is a fresh fake distinct from fakeWorkspaceFetchRepo.
|
// fakeWorktreeRepo is a fresh fake distinct from fakeWorkspaceFetchRepo.
|
||||||
type fakeWorktreeRepo struct {
|
type fakeWorktreeRepo struct {
|
||||||
warnList []runners.WorktreeRow
|
warnList []runners.WorktreeRow
|
||||||
@@ -172,6 +182,37 @@ func TestWorktreePrune_NoRowsIsNoOp(t *testing.T) {
|
|||||||
assert.Empty(t, disp.msgs)
|
assert.Empty(t, disp.msgs)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestWorktreePrune_AuditMetadataContainsWorkspaceIDAndPath(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
id := uuid.New()
|
||||||
|
workspaceID := uuid.New()
|
||||||
|
ownerID := uuid.New()
|
||||||
|
row := runners.WorktreeRow{
|
||||||
|
ID: id, WorkspaceID: workspaceID, OwnerID: ownerID,
|
||||||
|
Branch: "feat-audit", MainPath: "/main", Path: "/main/.worktrees/feat-audit",
|
||||||
|
}
|
||||||
|
|
||||||
|
repo := &fakeWorktreeRepo{
|
||||||
|
pruneList: []runners.WorktreeRow{row},
|
||||||
|
tryStartResult: row,
|
||||||
|
}
|
||||||
|
gitr := &fakePruner{}
|
||||||
|
rec := &recordingRecorder{}
|
||||||
|
cfg := runners.WorktreePruneConfig{IdleThreshold: 7 * 24 * time.Hour, WarningLead: 3 * 24 * time.Hour}
|
||||||
|
|
||||||
|
r := runners.NewWorktreePrune(repo, gitr, &recordingDispatcher{}, cfg, discardLogger(), rec)
|
||||||
|
r.Run(context.Background())
|
||||||
|
|
||||||
|
require.Len(t, rec.entries, 1)
|
||||||
|
e := rec.entries[0]
|
||||||
|
assert.Equal(t, "worktree.pruned", e.Action)
|
||||||
|
assert.Equal(t, "worktree", e.TargetType)
|
||||||
|
assert.Equal(t, id.String(), e.TargetID)
|
||||||
|
assert.Equal(t, workspaceID.String(), e.Metadata["workspace_id"])
|
||||||
|
assert.Equal(t, "feat-audit", e.Metadata["branch"])
|
||||||
|
assert.Equal(t, "/main/.worktrees/feat-audit", e.Metadata["path"])
|
||||||
|
}
|
||||||
|
|
||||||
func TestWorktreePrune_CASLostSkipsSilently(t *testing.T) {
|
func TestWorktreePrune_CASLostSkipsSilently(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
id := uuid.New()
|
id := uuid.New()
|
||||||
|
|||||||
@@ -35,12 +35,13 @@ RETURNING *;
|
|||||||
-- name: DeleteWorkspace :exec
|
-- name: DeleteWorkspace :exec
|
||||||
DELETE FROM workspaces WHERE id = $1;
|
DELETE FROM workspaces WHERE id = $1;
|
||||||
|
|
||||||
-- name: ResetStuckSyncStatuses :exec
|
-- name: ResetStuckSyncStatuses :many
|
||||||
UPDATE workspaces
|
UPDATE workspaces
|
||||||
SET sync_status = 'error',
|
SET sync_status = 'error',
|
||||||
last_sync_error = 'process_restarted_during_' || sync_status,
|
last_sync_error = 'process_restarted_during_' || sync_status,
|
||||||
updated_at = now()
|
updated_at = now()
|
||||||
WHERE sync_status IN ('cloning', 'syncing');
|
WHERE sync_status IN ('cloning', 'syncing')
|
||||||
|
RETURNING id, last_sync_error;
|
||||||
|
|
||||||
-- name: ListFetchTargets :many
|
-- name: ListFetchTargets :many
|
||||||
SELECT w.id, p.owner_id, w.name, w.main_path
|
SELECT w.id, p.owner_id, w.name, w.main_path
|
||||||
|
|||||||
@@ -44,7 +44,9 @@ type Repository interface {
|
|||||||
UpdateWorkspaceCore(ctx context.Context, id uuid.UUID, name, description, defaultBranch string) (*Workspace, error)
|
UpdateWorkspaceCore(ctx context.Context, id uuid.UUID, name, description, defaultBranch string) (*Workspace, error)
|
||||||
UpdateWorkspaceSyncStatus(ctx context.Context, id uuid.UUID, st SyncStatus, syncedAt *time.Time, errMsg string) (*Workspace, error)
|
UpdateWorkspaceSyncStatus(ctx context.Context, id uuid.UUID, st SyncStatus, syncedAt *time.Time, errMsg string) (*Workspace, error)
|
||||||
DeleteWorkspace(ctx context.Context, id uuid.UUID) error
|
DeleteWorkspace(ctx context.Context, id uuid.UUID) error
|
||||||
ResetStuckSyncStatuses(ctx context.Context) error
|
// ResetStuckSyncStatuses 把启动时残留的 cloning/syncing 复位为 error,
|
||||||
|
// 返回被复位的工作区 ID 列表(按发生顺序),由调用方写 audit。
|
||||||
|
ResetStuckSyncStatuses(ctx context.Context) ([]uuid.UUID, error)
|
||||||
|
|
||||||
// Worktree (no-tx 部分)
|
// Worktree (no-tx 部分)
|
||||||
GetWorktreeByID(ctx context.Context, id uuid.UUID) (*Worktree, error)
|
GetWorktreeByID(ctx context.Context, id uuid.UUID) (*Worktree, error)
|
||||||
@@ -240,11 +242,16 @@ func (r *pgRepo) DeleteWorkspace(ctx context.Context, id uuid.UUID) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (r *pgRepo) ResetStuckSyncStatuses(ctx context.Context) error {
|
func (r *pgRepo) ResetStuckSyncStatuses(ctx context.Context) ([]uuid.UUID, error) {
|
||||||
if err := r.q.ResetStuckSyncStatuses(ctx); err != nil {
|
rows, err := r.q.ResetStuckSyncStatuses(ctx)
|
||||||
return errs.Wrap(err, errs.CodeInternal, "reset stuck sync")
|
if err != nil {
|
||||||
|
return nil, errs.Wrap(err, errs.CodeInternal, "reset stuck sync")
|
||||||
}
|
}
|
||||||
return nil
|
ids := make([]uuid.UUID, 0, len(rows))
|
||||||
|
for _, row := range rows {
|
||||||
|
ids = append(ids, fromPgUUID(row.ID))
|
||||||
|
}
|
||||||
|
return ids, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// ===== Workspace fetch (jobs runner) =====
|
// ===== Workspace fetch (jobs runner) =====
|
||||||
|
|||||||
@@ -241,17 +241,38 @@ func (q *Queries) MarkSyncingCAS(ctx context.Context, id pgtype.UUID) (int64, er
|
|||||||
return result.RowsAffected(), nil
|
return result.RowsAffected(), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
const resetStuckSyncStatuses = `-- name: ResetStuckSyncStatuses :exec
|
const resetStuckSyncStatuses = `-- name: ResetStuckSyncStatuses :many
|
||||||
UPDATE workspaces
|
UPDATE workspaces
|
||||||
SET sync_status = 'error',
|
SET sync_status = 'error',
|
||||||
last_sync_error = 'process_restarted_during_' || sync_status,
|
last_sync_error = 'process_restarted_during_' || sync_status,
|
||||||
updated_at = now()
|
updated_at = now()
|
||||||
WHERE sync_status IN ('cloning', 'syncing')
|
WHERE sync_status IN ('cloning', 'syncing')
|
||||||
|
RETURNING id, last_sync_error
|
||||||
`
|
`
|
||||||
|
|
||||||
func (q *Queries) ResetStuckSyncStatuses(ctx context.Context) error {
|
type ResetStuckSyncStatusesRow struct {
|
||||||
_, err := q.db.Exec(ctx, resetStuckSyncStatuses)
|
ID pgtype.UUID `json:"id"`
|
||||||
return err
|
LastSyncError *string `json:"last_sync_error"`
|
||||||
|
}
|
||||||
|
|
||||||
|
func (q *Queries) ResetStuckSyncStatuses(ctx context.Context) ([]ResetStuckSyncStatusesRow, error) {
|
||||||
|
rows, err := q.db.Query(ctx, resetStuckSyncStatuses)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
var items []ResetStuckSyncStatusesRow
|
||||||
|
for rows.Next() {
|
||||||
|
var i ResetStuckSyncStatusesRow
|
||||||
|
if err := rows.Scan(&i.ID, &i.LastSyncError); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
items = append(items, i)
|
||||||
|
}
|
||||||
|
if err := rows.Err(); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return items, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
const updateWorkspaceCore = `-- name: UpdateWorkspaceCore :one
|
const updateWorkspaceCore = `-- name: UpdateWorkspaceCore :one
|
||||||
|
|||||||
@@ -75,7 +75,9 @@ func (f *fakeRepo) DeleteWorkspace(_ context.Context, id uuid.UUID) error {
|
|||||||
delete(f.wss, id)
|
delete(f.wss, id)
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
func (f *fakeRepo) ResetStuckSyncStatuses(_ context.Context) error { return nil }
|
func (f *fakeRepo) ResetStuckSyncStatuses(_ context.Context) ([]uuid.UUID, error) {
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
func (f *fakeRepo) ListFetchTargets(_ context.Context) ([]FetchTarget, error) { return nil, nil }
|
func (f *fakeRepo) ListFetchTargets(_ context.Context) ([]FetchTarget, error) { return nil, nil }
|
||||||
func (f *fakeRepo) MarkSyncing(_ context.Context, _ uuid.UUID) (bool, error) { return false, nil }
|
func (f *fakeRepo) MarkSyncing(_ context.Context, _ uuid.UUID) (bool, error) { return false, nil }
|
||||||
func (f *fakeRepo) MarkFetchResult(_ context.Context, _ uuid.UUID, _ bool, _ string) error {
|
func (f *fakeRepo) MarkFetchResult(_ context.Context, _ uuid.UUID, _ bool, _ string) error {
|
||||||
|
|||||||
Reference in New Issue
Block a user