返回 DeepSeek-Reasonix
mutation_lock_acquire.go
根目录 / internal / repair / mutation_lock_acquire.go
1 package repair
2
3 import (
4 "context"
5 "crypto/sha256"
6 "errors"
7 "fmt"
8 "os"
9 "path/filepath"
10 "slices"
11 "sort"
12 "strings"
13 "time"
14
15 "reasonix/internal/filelock"
16 "reasonix/internal/pathidentity"
17 )
18
19 type repairMutationTarget struct {
20 path string
21 key string
22 info os.FileInfo
23 exists bool
24 link string
25 }
26
27 type repairMutationLockDomain struct {
28 path string
29 key string
30 }
31
32 func repairMutationTargets(paths []string) ([]repairMutationTarget, []string, []string, error) {
33 targets := make([]repairMutationTarget, 0, len(paths))
34 primary := map[string]struct{}{}
35 locks := map[string]struct{}{}
36 for _, path := range paths {
37 path = strings.TrimSpace(path)
38 if path == "" {
39 continue
40 }
41 identity, err := pathidentity.Resolve(path, pathidentity.Options{FollowLeaf: false})
42 if err != nil {
43 return nil, nil, nil, fmt.Errorf("lock repair mutations: resolve target: %w", err)
44 }
45 if _, exists := primary[identity.Key]; exists {
46 continue
47 }
48 info, link, statErr := inspectRepairMutationEntry(identity.AccessPath)
49 exists := statErr == nil
50 if statErr != nil && !os.IsNotExist(statErr) {
51 return nil, nil, nil, fmt.Errorf("lock repair mutations: inspect target: %w", statErr)
52 }
53 primary[identity.Key] = struct{}{}
54 targets = append(targets, repairMutationTarget{identity.AccessPath, identity.Key, info, exists, link})
55 locks[identity.Key] = struct{}{}
56 if legacy := legacyCanonicalRepairPath(path); legacy != "" {
57 locks[legacy] = struct{}{}
58 }
59 }
60 return targets, sortedRepairKeys(primary), sortedRepairKeys(locks), nil
61 }
62
63 func sortedRepairKeys(set map[string]struct{}) []string {
64 keys := make([]string, 0, len(set))
65 for key := range set {
66 keys = append(keys, key)
67 }
68 sort.Strings(keys)
69 return keys
70 }
71
72 func repairMutationLockDomains(lockDir string, keys []string) ([]repairMutationLockDomain, error) {
73 domains := make([]repairMutationLockDomain, 0, len(keys))
74 for _, key := range keys {
75 digest := sha256.Sum256([]byte(key))
76 path := filepath.Join(lockDir, fmt.Sprintf("%x.lock", digest))
77 identity, err := pathidentity.Resolve(path, pathidentity.Options{FollowLeaf: false})
78 if err != nil {
79 return nil, fmt.Errorf("lock repair mutations: resolve lock identity: %w", err)
80 }
81 domains = append(domains, repairMutationLockDomain{path: path, key: identity.Key})
82 }
83 sort.Slice(domains, func(i, j int) bool { return domains[i].key < domains[j].key })
84 return domains, nil
85 }
86
87 func acquireRepairMutationLocks(timeout time.Duration, domains []repairMutationLockDomain) ([]func(), error) {
88 if timeout <= 0 {
89 timeout = repairMutationLockTimeout
90 }
91 ctx, cancel := context.WithTimeout(context.Background(), timeout)
92 defer cancel()
93 for {
94 releases := make([]func(), 0, len(domains))
95 for _, domain := range domains {
96 release, err := filelock.TryAcquireModeWithKey(domain.path, domain.key, filelock.ModeExclusive)
97 if err == nil {
98 releases = append(releases, release)
99 continue
100 }
101 releaseRepairMutationLocks(releases)
102 if !errors.Is(err, filelock.ErrHeld) {
103 return nil, fmt.Errorf("lock repair mutations: %w", err)
104 }
105 break
106 }
107 if len(releases) == len(domains) {
108 return releases, nil
109 }
110 timer := time.NewTimer(20 * time.Millisecond)
111 select {
112 case <-timer.C:
113 case <-ctx.Done():
114 if !timer.Stop() {
115 select {
116 case <-timer.C:
117 default:
118 }
119 }
120 return nil, fmt.Errorf("lock repair mutations: %w", ctx.Err())
121 }
122 }
123 }
124
125 func revalidateRepairMutationTargets(targets []repairMutationTarget) error {
126 for _, target := range targets {
127 identity, resolveErr := pathidentity.Resolve(target.path, pathidentity.Options{FollowLeaf: false})
128 currentInfo, currentLink, statErr := inspectRepairMutationEntry(target.path)
129 currentExists := statErr == nil
130 if statErr != nil && !os.IsNotExist(statErr) && resolveErr == nil {
131 resolveErr = statErr
132 }
133 if resolveErr != nil {
134 return fmt.Errorf("lock repair mutations: revalidate target: %w", resolveErr)
135 }
136 if identity.Key != target.key || currentLink != target.link || repairEntryRedirected(target.info, target.exists, currentInfo, currentExists) {
137 return errors.New("lock repair mutations: target identity changed while waiting")
138 }
139 }
140 return nil
141 }
142
143 func inspectRepairMutationEntry(path string) (os.FileInfo, string, error) {
144 info, err := os.Lstat(path)
145 if err != nil {
146 return nil, "", err
147 }
148 if info.Mode()&os.ModeSymlink != 0 {
149 // Inodes can be reused immediately after unlink. Preserve the link
150 // destination as well so a redirected link cannot pass SameFile.
151 link, err := os.Readlink(path)
152 return info, link, err
153 }
154 return info, "", nil
155 }
156
157 func releaseRepairMutationLocks(releases []func()) {
158 for _, release := range slices.Backward(releases) {
159 release()
160 }
161 }
162
163 // Regular files may be atomically replaced under the lock; content checks
164 // detect stale state. Other node types retain native identity.
165 func repairEntryRedirected(before os.FileInfo, beforeExists bool, after os.FileInfo, afterExists bool) bool {
166 if (!beforeExists || before.Mode().IsRegular()) && (!afterExists || after.Mode().IsRegular()) {
167 return false
168 }
169 return beforeExists != afterExists || !os.SameFile(before, after)
170 }
171
171 lines GO