feat(workspace): add WorkspaceRef fencing and hide physical paths
Cross-module callers get an opaque ref plus rotating lease tokens so stale writers fail closed, and file I/O is refused inside a unit of work. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -0,0 +1,8 @@
|
||||
-- +goose Up
|
||||
ALTER TABLE document_records
|
||||
ADD COLUMN IF NOT EXISTS lease_fencing_token TEXT NOT NULL DEFAULT '',
|
||||
ADD COLUMN IF NOT EXISTS workspace_schema_version INT NOT NULL DEFAULT 1;
|
||||
|
||||
-- +goose Down
|
||||
ALTER TABLE document_records DROP COLUMN IF EXISTS workspace_schema_version;
|
||||
ALTER TABLE document_records DROP COLUMN IF EXISTS lease_fencing_token;
|
||||
@@ -2,8 +2,11 @@ package docgov
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
@@ -29,6 +32,7 @@ type ContentSaveRequest struct {
|
||||
ActorName string
|
||||
MarkdownBody string
|
||||
ExpectedSHA256 string // 乐观并发:非空时必须匹配当前 ContentSHA256
|
||||
FencingToken string
|
||||
}
|
||||
|
||||
// ContentSaveResult 正文保存结果
|
||||
@@ -101,6 +105,7 @@ func (e *GovernanceEngine) AcquireLease(ctx context.Context, req LeaseAcquireReq
|
||||
}
|
||||
doc.LeaseExpiresAt = &expires
|
||||
doc.LeaseHeartbeatAt = &now
|
||||
doc.LeaseFencingToken = newLeaseFencingToken()
|
||||
doc.UpdatedAt = now
|
||||
|
||||
e.persistLeaseLocked(ctx, doc)
|
||||
@@ -344,6 +349,11 @@ func (e *GovernanceEngine) SaveDocumentContent(ctx context.Context, req ContentS
|
||||
if req.ActorID <= 0 {
|
||||
holderID = "human_anon"
|
||||
}
|
||||
if req.FencingToken != "" {
|
||||
if err := e.AssertLeaseFencing(ctx, req.DocID, req.FencingToken); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
doc, err := e.AcquireLease(ctx, LeaseAcquireRequest{
|
||||
DocID: req.DocID,
|
||||
@@ -409,6 +419,28 @@ func (e *GovernanceEngine) SaveDocumentContent(ctx context.Context, req ContentS
|
||||
return &ContentSaveResult{Doc: doc, ContentSHA256: shaStr, WorkDir: workDir}, nil
|
||||
}
|
||||
|
||||
// AssertLeaseFencing rejects writers holding a superseded token.
|
||||
func (e *GovernanceEngine) AssertLeaseFencing(ctx context.Context, docID, token string) error {
|
||||
e.mu.Lock()
|
||||
defer e.mu.Unlock()
|
||||
doc, err := e.getDocLocked(ctx, docID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if token == "" || doc.LeaseFencingToken == "" || token != doc.LeaseFencingToken {
|
||||
return ErrStaleFencing
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func newLeaseFencingToken() string {
|
||||
var b [16]byte
|
||||
if _, err := io.ReadFull(rand.Reader, b[:]); err != nil {
|
||||
return fmt.Sprintf("%d", time.Now().UnixNano())
|
||||
}
|
||||
return hex.EncodeToString(b[:])
|
||||
}
|
||||
|
||||
func (e *GovernanceEngine) getDocLocked(ctx context.Context, docID string) (*DocumentRecord, error) {
|
||||
doc, exists := e.memDocs[docID]
|
||||
if !exists {
|
||||
@@ -428,10 +460,10 @@ func (e *GovernanceEngine) persistLeaseLocked(ctx context.Context, doc *Document
|
||||
_, _ = e.pool.Exec(ctx, `
|
||||
UPDATE document_records
|
||||
SET lease_holder_type = $1, lease_holder_id = $2, lease_work_dir = $3,
|
||||
lease_expires_at = $4, lease_heartbeat_at = $5, updated_at = $6
|
||||
WHERE doc_id = $7
|
||||
lease_expires_at = $4, lease_heartbeat_at = $5, lease_fencing_token = $6, updated_at = $7
|
||||
WHERE doc_id = $8
|
||||
`, string(doc.LeaseHolderType), doc.LeaseHolderID, doc.LeaseWorkDir,
|
||||
doc.LeaseExpiresAt, doc.LeaseHeartbeatAt, doc.UpdatedAt, doc.DocID)
|
||||
doc.LeaseExpiresAt, doc.LeaseHeartbeatAt, doc.LeaseFencingToken, doc.UpdatedAt, doc.DocID)
|
||||
}
|
||||
|
||||
func (e *GovernanceEngine) appendAuditLocked(doc *DocumentRecord, action string, actorID int64, actorName string, actorRole UserRole, details map[string]any) {
|
||||
|
||||
@@ -86,6 +86,38 @@ func TestSameHolderRenewLease(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestStaleFencingRejectedAfterRenew(t *testing.T) {
|
||||
engine := docgov.NewGovernanceEngine(nil)
|
||||
ctx := context.Background()
|
||||
docID := "doc_fence_001"
|
||||
_, _ = engine.CreateOrGetDocument(ctx, docID, 1, "intake", "摸底", "V0.1", "", 1, "a", 2, "b", 3, "c")
|
||||
first, err := engine.AcquireLease(ctx, docgov.LeaseAcquireRequest{
|
||||
DocID: docID, HolderType: docgov.LeaseHolderHuman, HolderID: "1", TTL: time.Minute,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
old := first.LeaseFencingToken
|
||||
if old == "" {
|
||||
t.Fatal("expected fencing token")
|
||||
}
|
||||
second, err := engine.AcquireLease(ctx, docgov.LeaseAcquireRequest{
|
||||
DocID: docID, HolderType: docgov.LeaseHolderHuman, HolderID: "1", TTL: time.Minute,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if second.LeaseFencingToken == old {
|
||||
t.Fatal("renew must rotate fencing token")
|
||||
}
|
||||
if err := engine.AssertLeaseFencing(ctx, docID, old); err != docgov.ErrStaleFencing {
|
||||
t.Fatalf("old writer: %v", err)
|
||||
}
|
||||
if err := engine.AssertLeaseFencing(ctx, docID, second.LeaseFencingToken); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAssertHumanCanEdit_ChecksDutyBeforeLease(t *testing.T) {
|
||||
engine := docgov.NewGovernanceEngine(nil)
|
||||
ctx := context.Background()
|
||||
|
||||
@@ -18,6 +18,7 @@ var (
|
||||
ErrProposalAlreadyDecided = errors.New("docgov: change proposal has already been decided")
|
||||
ErrChangeProposalPending = errors.New("docgov: document is claimed and has pending change proposal waiting for author approval")
|
||||
ErrDocumentBusy = errors.New("docgov: document edit lease is held by another writer")
|
||||
ErrStaleFencing = errors.New("docgov: stale lease fencing token")
|
||||
ErrLeaseExpired = errors.New("docgov: document edit lease has expired")
|
||||
ErrLeaseNotHeld = errors.New("docgov: caller does not hold the document edit lease")
|
||||
ErrForceEditNotAllowed = errors.New("docgov: force_direct_edit requires an approved_by_author proposal and unlocked drafting document")
|
||||
@@ -183,11 +184,12 @@ type DocumentRecord struct {
|
||||
ApprovedAt *time.Time `json:"approved_at"`
|
||||
|
||||
// 编辑租约 (运行时互斥,正交于 IsLocked)
|
||||
LeaseHolderType LeaseHolderType `json:"lease_holder_type,omitempty"`
|
||||
LeaseHolderID string `json:"lease_holder_id,omitempty"`
|
||||
LeaseWorkDir string `json:"lease_work_dir,omitempty"`
|
||||
LeaseExpiresAt *time.Time `json:"lease_expires_at,omitempty"`
|
||||
LeaseHeartbeatAt *time.Time `json:"lease_heartbeat_at,omitempty"`
|
||||
LeaseHolderType LeaseHolderType `json:"lease_holder_type,omitempty"`
|
||||
LeaseHolderID string `json:"lease_holder_id,omitempty"`
|
||||
LeaseWorkDir string `json:"lease_work_dir,omitempty"`
|
||||
LeaseExpiresAt *time.Time `json:"lease_expires_at,omitempty"`
|
||||
LeaseHeartbeatAt *time.Time `json:"lease_heartbeat_at,omitempty"`
|
||||
LeaseFencingToken string `json:"lease_fencing_token,omitempty"`
|
||||
|
||||
Config DocumentConfig `json:"config"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
@@ -212,6 +214,7 @@ func (d *DocumentRecord) ClearLease() {
|
||||
d.LeaseWorkDir = ""
|
||||
d.LeaseExpiresAt = nil
|
||||
d.LeaseHeartbeatAt = nil
|
||||
d.LeaseFencingToken = ""
|
||||
}
|
||||
|
||||
// CommentType 批注类型
|
||||
|
||||
@@ -0,0 +1,15 @@
|
||||
package squad
|
||||
|
||||
import "chat-agent-system/internal/workspace"
|
||||
|
||||
// WorkspaceRefForRun is the only squad→workspace DTO. Physical paths stay inside workspace.
|
||||
func WorkspaceRefForRun(runID string, projectID int64, env, token string) workspace.WorkspaceRef {
|
||||
return workspace.WorkspaceRef{
|
||||
ID: "run-" + runID,
|
||||
SchemaVersion: workspace.SchemaVersion,
|
||||
FencingToken: token,
|
||||
ProjectScope: projectID,
|
||||
RunID: runID,
|
||||
Environment: env,
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,219 @@
|
||||
package workspace
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"syscall"
|
||||
|
||||
"chat-agent-system/internal/app/uow"
|
||||
)
|
||||
|
||||
// Capability is the only module allowed to turn a WorkspaceRef into bytes on disk.
|
||||
type Capability struct {
|
||||
dataRoot string
|
||||
leases *LeaseTable
|
||||
}
|
||||
|
||||
func NewCapability(dataRoot string, leases *LeaseTable) *Capability {
|
||||
if leases == nil {
|
||||
leases = NewLeaseTable()
|
||||
}
|
||||
return &Capability{dataRoot: dataRoot, leases: leases}
|
||||
}
|
||||
|
||||
func (c *Capability) Issue(ref WorkspaceRef) (WorkspaceRef, error) {
|
||||
if err := c.validateRefMeta(ref); err != nil {
|
||||
return WorkspaceRef{}, err
|
||||
}
|
||||
token, err := newFencingToken()
|
||||
if err != nil {
|
||||
return WorkspaceRef{}, err
|
||||
}
|
||||
issued := ref
|
||||
issued.SchemaVersion = SchemaVersion
|
||||
issued.FencingToken = token
|
||||
c.leases.Issue(issued)
|
||||
return issued, nil
|
||||
}
|
||||
|
||||
func (c *Capability) ReadArtifact(ctx context.Context, ref WorkspaceRef, key string) ([]byte, error) {
|
||||
if err := c.guard(ctx, ref, false); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
path, err := c.resolveKey(ref, key)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
resolved, err := c.enforceRealpath(path)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return os.ReadFile(resolved)
|
||||
}
|
||||
|
||||
func (c *Capability) WriteArtifact(ctx context.Context, ref WorkspaceRef, key string, body []byte) error {
|
||||
if err := c.guard(ctx, ref, true); err != nil {
|
||||
return err
|
||||
}
|
||||
path, err := c.resolveKey(ref, key)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
|
||||
return err
|
||||
}
|
||||
tmp := path + ".tmp-" + ref.FencingToken[:8]
|
||||
if err := os.WriteFile(tmp, body, 0o644); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := c.enforceRealpath(tmp); err != nil {
|
||||
_ = os.Remove(tmp)
|
||||
return err
|
||||
}
|
||||
if err := os.Rename(tmp, path); err != nil {
|
||||
_ = os.Remove(tmp)
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Capability) CommitProjection(ctx context.Context, ref WorkspaceRef, key, digest string) error {
|
||||
if err := c.guard(ctx, ref, true); err != nil {
|
||||
return err
|
||||
}
|
||||
body, err := c.ReadArtifact(ctx, ref, key)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
sum := sha256.Sum256(body)
|
||||
got := hex.EncodeToString(sum[:])
|
||||
if digest != "" && digest != got {
|
||||
return fmt.Errorf("workspace: digest mismatch")
|
||||
}
|
||||
marker := []byte(got + "\ncomplete\n")
|
||||
return c.WriteArtifact(ctx, ref, key+".complete", marker)
|
||||
}
|
||||
|
||||
func (c *Capability) guard(ctx context.Context, ref WorkspaceRef, write bool) error {
|
||||
if ctx != nil {
|
||||
if _, ok := uow.ScopeFrom(ctx); ok {
|
||||
return ErrIOInTransaction
|
||||
}
|
||||
}
|
||||
if err := c.validateRefMeta(ref); err != nil {
|
||||
return err
|
||||
}
|
||||
if write {
|
||||
return c.leases.Assert(ref)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Capability) validateRefMeta(ref WorkspaceRef) error {
|
||||
if strings.TrimSpace(ref.ID) == "" {
|
||||
return ErrInvalidKey
|
||||
}
|
||||
if ref.SchemaVersion != 0 && ref.SchemaVersion != SchemaVersion {
|
||||
return ErrUnknownSchema
|
||||
}
|
||||
if ref.ProjectScope < 0 {
|
||||
return ErrCrossProject
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Capability) resolveKey(ref WorkspaceRef, key string) (string, error) {
|
||||
if err := validateArtifactKey(key); err != nil {
|
||||
return "", err
|
||||
}
|
||||
root := filepath.Join(c.dataRoot, "workspaces", ref.ID)
|
||||
joined := filepath.Join(root, filepath.FromSlash(key))
|
||||
absRoot, err := filepath.Abs(root)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
absJoin, err := filepath.Abs(joined)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
rel, err := filepath.Rel(absRoot, absJoin)
|
||||
if err != nil || rel == ".." || strings.HasPrefix(rel, ".."+string(os.PathSeparator)) {
|
||||
return "", ErrPathEscape
|
||||
}
|
||||
return absJoin, nil
|
||||
}
|
||||
|
||||
func (c *Capability) enforceRealpath(path string) (string, error) {
|
||||
info, err := os.Lstat(path)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
return path, nil
|
||||
}
|
||||
return "", err
|
||||
}
|
||||
if info.Mode()&os.ModeSymlink != 0 {
|
||||
return "", ErrPathEscape
|
||||
}
|
||||
if info.Mode()&os.ModeNamedPipe != 0 {
|
||||
return "", ErrPathEscape
|
||||
}
|
||||
resolved, err := filepath.EvalSymlinks(filepath.Dir(path))
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
full := filepath.Join(resolved, filepath.Base(path))
|
||||
root, err := filepath.Abs(c.dataRoot)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
rel, err := filepath.Rel(root, full)
|
||||
if err != nil || rel == ".." || strings.HasPrefix(rel, ".."+string(os.PathSeparator)) {
|
||||
return "", ErrPathEscape
|
||||
}
|
||||
if st, ok := info.Sys().(*syscall.Stat_t); ok && st.Nlink > 1 && !info.IsDir() {
|
||||
// Hardlink to an inode that may live outside: still require the named path stay in root.
|
||||
if !strings.HasPrefix(full, root) {
|
||||
return "", ErrPathEscape
|
||||
}
|
||||
}
|
||||
return full, nil
|
||||
}
|
||||
|
||||
func validateArtifactKey(key string) error {
|
||||
key = strings.TrimSpace(key)
|
||||
if key == "" {
|
||||
return ErrInvalidKey
|
||||
}
|
||||
if strings.ContainsRune(key, 0) {
|
||||
return ErrInvalidKey
|
||||
}
|
||||
if filepath.IsAbs(key) || strings.HasPrefix(key, "/") || strings.HasPrefix(key, `\`) {
|
||||
return ErrInvalidKey
|
||||
}
|
||||
lower := strings.ToLower(key)
|
||||
if strings.Contains(lower, "%2e%2e") || strings.Contains(lower, "%2f") || strings.Contains(lower, "%5c") {
|
||||
return ErrInvalidKey
|
||||
}
|
||||
cleaned := filepath.ToSlash(key)
|
||||
for _, part := range strings.Split(cleaned, "/") {
|
||||
if part == "" || part == "." || part == ".." {
|
||||
return ErrInvalidKey
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func newFencingToken() (string, error) {
|
||||
var b [16]byte
|
||||
if _, err := io.ReadFull(rand.Reader, b[:]); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return hex.EncodeToString(b[:]), nil
|
||||
}
|
||||
@@ -0,0 +1,119 @@
|
||||
package workspace
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"chat-agent-system/internal/app/uow"
|
||||
)
|
||||
|
||||
func issued(t *testing.T, cap *Capability, id string, project int64) WorkspaceRef {
|
||||
t.Helper()
|
||||
ref, err := cap.Issue(WorkspaceRef{ID: id, ProjectScope: project, SchemaVersion: SchemaVersion})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return ref
|
||||
}
|
||||
|
||||
func TestWorkspaceRefHasNoPhysicalPathFields(t *testing.T) {
|
||||
var ref WorkspaceRef
|
||||
_ = ref.ID
|
||||
_ = ref.SchemaVersion
|
||||
_ = ref.FencingToken
|
||||
_ = ref.ProjectScope
|
||||
_ = ref.RunID
|
||||
_ = ref.Environment
|
||||
}
|
||||
|
||||
func TestCapabilityRejectsBadKeys(t *testing.T) {
|
||||
cap := NewCapability(t.TempDir(), nil)
|
||||
ref := issued(t, cap, "ws1", 7)
|
||||
ctx := context.Background()
|
||||
for _, key := range []string{"", "../etc/passwd", "/etc/passwd", "..", "a/../b", "a//b", "%2e%2e/secret", "ok/%2fetc"} {
|
||||
if _, err := cap.ReadArtifact(ctx, ref, key); err == nil {
|
||||
t.Fatalf("key %q should reject", key)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestCapabilityRejectsSymlinkEscape(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
cap := NewCapability(root, nil)
|
||||
ref := issued(t, cap, "ws1", 7)
|
||||
outside := filepath.Join(t.TempDir(), "secret")
|
||||
if err := os.WriteFile(outside, []byte("nope"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
wsDir := filepath.Join(root, "workspaces", "ws1")
|
||||
if err := os.MkdirAll(wsDir, 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
link := filepath.Join(wsDir, "leak")
|
||||
if err := os.Symlink(outside, link); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := cap.ReadArtifact(context.Background(), ref, "leak"); err != ErrPathEscape {
|
||||
t.Fatalf("symlink: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStaleFencingRejected(t *testing.T) {
|
||||
cap := NewCapability(t.TempDir(), nil)
|
||||
old := issued(t, cap, "ws1", 3)
|
||||
_ = issued(t, cap, "ws1", 3)
|
||||
ctx := context.Background()
|
||||
if err := cap.WriteArtifact(ctx, old, "output.md", []byte("stale")); err != ErrStaleFencing {
|
||||
t.Fatalf("old writer: %v", err)
|
||||
}
|
||||
fresh := issued(t, cap, "ws1", 3)
|
||||
if err := cap.WriteArtifact(ctx, fresh, "output.md", []byte("ok")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestUnknownSchemaRejected(t *testing.T) {
|
||||
cap := NewCapability(t.TempDir(), nil)
|
||||
_, err := cap.Issue(WorkspaceRef{ID: "ws1", SchemaVersion: 99})
|
||||
if err != ErrUnknownSchema {
|
||||
t.Fatalf("got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNoFileIOInsideUnitOfWork(t *testing.T) {
|
||||
cap := NewCapability(t.TempDir(), nil)
|
||||
ref := issued(t, cap, "ws1", 1)
|
||||
ctx := uow.WithScope(context.Background(), stubScope{})
|
||||
if err := cap.WriteArtifact(ctx, ref, "output.md", []byte("x")); err != ErrIOInTransaction {
|
||||
t.Fatalf("got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCommitProjectionCompleteMarker(t *testing.T) {
|
||||
cap := NewCapability(t.TempDir(), nil)
|
||||
ref := issued(t, cap, "ws1", 1)
|
||||
ctx := context.Background()
|
||||
if err := cap.WriteArtifact(ctx, ref, "output.md", []byte("hello")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := cap.CommitProjection(ctx, ref, "output.md", ""); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
got, err := cap.ReadArtifact(ctx, ref, "output.md.complete")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !containsComplete(got) {
|
||||
t.Fatalf("marker %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
func containsComplete(b []byte) bool {
|
||||
return string(b) != "" && (string(b)[len(b)-1] == '\n')
|
||||
}
|
||||
|
||||
type stubScope struct{}
|
||||
|
||||
func (stubScope) Context() context.Context { return context.Background() }
|
||||
@@ -0,0 +1,35 @@
|
||||
package workspace
|
||||
|
||||
import "sync"
|
||||
|
||||
// LeaseTable stores the current fencing token per workspace ID.
|
||||
// Cross-replica correctness is the token itself, not this process mutex.
|
||||
type LeaseTable struct {
|
||||
mu sync.Mutex
|
||||
tokens map[string]string
|
||||
owner map[string]int64
|
||||
}
|
||||
|
||||
func NewLeaseTable() *LeaseTable {
|
||||
return &LeaseTable{tokens: map[string]string{}, owner: map[string]int64{}}
|
||||
}
|
||||
|
||||
func (t *LeaseTable) Issue(ref WorkspaceRef) {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
t.tokens[ref.ID] = ref.FencingToken
|
||||
t.owner[ref.ID] = ref.ProjectScope
|
||||
}
|
||||
|
||||
func (t *LeaseTable) Assert(ref WorkspaceRef) error {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
cur, ok := t.tokens[ref.ID]
|
||||
if !ok || cur == "" || ref.FencingToken == "" || cur != ref.FencingToken {
|
||||
return ErrStaleFencing
|
||||
}
|
||||
if scope, ok := t.owner[ref.ID]; ok && ref.ProjectScope != 0 && scope != 0 && scope != ref.ProjectScope {
|
||||
return ErrCrossProject
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,17 @@
|
||||
package workspace
|
||||
|
||||
import "testing"
|
||||
|
||||
func TestLeaseTableRejectsStaleTokenAfterReissue(t *testing.T) {
|
||||
table := NewLeaseTable()
|
||||
old := WorkspaceRef{ID: "ws1", FencingToken: "old-token", ProjectScope: 3}
|
||||
table.Issue(old)
|
||||
fresh := WorkspaceRef{ID: "ws1", FencingToken: "new-token", ProjectScope: 3}
|
||||
table.Issue(fresh)
|
||||
if err := table.Assert(old); err != ErrStaleFencing {
|
||||
t.Fatalf("stale: %v", err)
|
||||
}
|
||||
if err := table.Assert(fresh); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
@@ -2,7 +2,6 @@ package workspace
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
@@ -11,12 +10,7 @@ import (
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrPathEscape = errors.New("workspace: path escapes data root")
|
||||
ErrWorkDirNotFound = errors.New("workspace: work directory not found")
|
||||
)
|
||||
|
||||
// Resolver 将逻辑 ID 解析为受约束的工作区路径。
|
||||
// Resolver 将逻辑 ID 解析为受约束的工作区路径。物理路径不得进入 WorkspaceRef。
|
||||
type Resolver struct {
|
||||
dataRoot string
|
||||
pool *pgxpool.Pool
|
||||
|
||||
@@ -0,0 +1,28 @@
|
||||
package workspace
|
||||
|
||||
import "errors"
|
||||
|
||||
const SchemaVersion = 1
|
||||
|
||||
var (
|
||||
ErrInvalidKey = errors.New("workspace: invalid artifact key")
|
||||
ErrPathEscape = errors.New("workspace: path escapes workspace")
|
||||
ErrStaleFencing = errors.New("workspace: stale fencing token")
|
||||
ErrUnknownSchema = errors.New("workspace: unknown schema version")
|
||||
ErrIOInTransaction = errors.New("workspace: file I/O is forbidden inside a unit of work")
|
||||
ErrCrossProject = errors.New("workspace: workspace id is not in project scope")
|
||||
ErrLegacyReadOnly = errors.New("workspace: legacy-v0 workspace is read-only")
|
||||
ErrWorkDirNotFound = errors.New("workspace: work directory not found")
|
||||
ErrIncompleteMarker = errors.New("workspace: projection marker is incomplete")
|
||||
)
|
||||
|
||||
// WorkspaceRef is the only cross-module workspace DTO. It must not carry
|
||||
// physical paths (no Root, OutputDir, or ManifestPath).
|
||||
type WorkspaceRef struct {
|
||||
ID string
|
||||
SchemaVersion int
|
||||
FencingToken string
|
||||
ProjectScope int64
|
||||
RunID string
|
||||
Environment string
|
||||
}
|
||||
Reference in New Issue
Block a user