返回 DeepSeek-Reasonix
idempotency_store.go
根目录 / internal / sessioninbox / idempotency_store.go
1 package sessioninbox
2
3 import "time"
4
5 // LookupEnvelopeReceipt checks semantic identity before sources are read again.
6 func (s *Store) LookupEnvelopeReceipt(key string, env PromptEnvelope) (InboxReceipt, bool, error) {
7 if key == "" {
8 return InboxReceipt{}, false, nil
9 }
10 hash, err := idempotencyRequestHash(env)
11 if err != nil {
12 return InboxReceipt{}, false, err
13 }
14 s.mu.Lock()
15 defer s.mu.Unlock()
16 release, err := s.beginDiskTransactionLocked()
17 if err != nil {
18 return InboxReceipt{}, false, err
19 }
20 defer release()
21 return s.idempotentReceiptLocked(key, hash)
22 }
23
24 // LookupReceipt reads the existing bounded idempotency records without
25 // creating or replaying a write. Used after an uncertain transport outcome.
26 func (s *Store) LookupReceipt(key string) (InboxReceipt, bool) {
27 s.mu.Lock()
28 defer s.mu.Unlock()
29 if key == "" {
30 return InboxReceipt{}, false
31 }
32 if id, ok := s.man.Idempotency[key]; ok {
33 if _, found := s.man.item(id); found {
34 return InboxReceipt{ItemID: id, Disposition: DispositionIdempotentHit, Position: s.man.indexOf(id) + 1, Paused: s.man.Paused, Idempotent: true}, true
35 }
36 }
37 if receipt, ok := s.man.Receipts[key]; ok && time.Since(receipt.CompletedAt) <= idempotencyReceiptTTL {
38 return InboxReceipt{ItemID: receipt.ItemID, Disposition: DispositionIdempotentHit, Paused: s.man.Paused, Idempotent: true}, true
39 }
40 return InboxReceipt{}, false
41 }
42
43 func (s *Store) idempotentReceiptLocked(key, requestHash string) (InboxReceipt, bool, error) {
44 if key == "" {
45 return InboxReceipt{}, false, nil
46 }
47 if id, ok := s.man.Idempotency[key]; ok {
48 if item, found := s.man.item(id); found {
49 if previous := s.man.IdempotencyHashes[key]; previous != "" && previous != requestHash {
50 return InboxReceipt{}, false, ErrIdempotencyConflict
51 }
52 return InboxReceipt{
53 ItemID: item.ID, Disposition: DispositionIdempotentHit,
54 Position: s.man.indexOf(item.ID) + 1, Paused: s.man.Paused,
55 Capacity: s.snapshotLocked().Capacity, Idempotent: true,
56 }, true, nil
57 }
58 }
59 receipt, ok := s.man.Receipts[key]
60 if !ok || time.Since(receipt.CompletedAt) > idempotencyReceiptTTL {
61 return InboxReceipt{}, false, nil
62 }
63 if receipt.RequestHash != requestHash {
64 return InboxReceipt{}, false, ErrIdempotencyConflict
65 }
66 return InboxReceipt{
67 ItemID: receipt.ItemID, Disposition: DispositionIdempotentHit,
68 Paused: s.man.Paused, Capacity: s.snapshotLocked().Capacity, Idempotent: true,
69 }, true, nil
70 }
71
72 func (s *Store) idempotentAliasReplayLocked(key, requestHash, itemID string) (bool, error) {
73 if key == "" {
74 return false, nil
75 }
76 if existingID, ok := s.man.Idempotency[key]; ok {
77 if existingHash := s.man.IdempotencyHashes[key]; existingHash != "" && existingHash != requestHash {
78 return false, ErrIdempotencyConflict
79 }
80 if existingID != itemID {
81 return false, ErrIdempotencyConflict
82 }
83 return true, nil
84 }
85 receipt, ok := s.man.Receipts[key]
86 if !ok || time.Since(receipt.CompletedAt) > idempotencyReceiptTTL {
87 return false, nil
88 }
89 if receipt.RequestHash != requestHash || receipt.ItemID != itemID {
90 return false, ErrIdempotencyConflict
91 }
92 return true, nil
93 }
94
95 func bindIdempotency(m *manifest, key, itemID, requestHash string) {
96 if key == "" {
97 return
98 }
99 if m.Idempotency == nil {
100 m.Idempotency = map[string]string{}
101 }
102 if m.IdempotencyHashes == nil {
103 m.IdempotencyHashes = map[string]string{}
104 }
105 m.Idempotency[key] = itemID
106 m.IdempotencyHashes[key] = requestHash
107 }
108
108 lines GO