| 1 | package ledger |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "os" |
| 9 | "time" |
| 10 | |
| 11 | "github.com/carlosframework/platform/internal/release" |
| 12 | ) |
| 13 | |
| 14 | type Appended struct { |
| 15 | Key string `json:"key"` |
| 16 | SHA256 string `json:"sha256"` |
| 17 | Count int `json:"count"` |
| 18 | } |
| 19 | |
| 20 | // appendAttempts bounds the CAS retry loop. Conflicts are rare (two |
| 21 | // writers accruing the same app's ledger in the same moment); four |
| 22 | // attempts outlasts any plausible burst without ever spinning. |
| 23 | const appendAttempts = 4 |
| 24 | |
| 25 | // ErrBusy marks the one Append failure that is neither the caller's fault |
| 26 | // nor a broken store: too many writers accruing the same chain at once, so |
| 27 | // every CAS attempt lost. A caller that retries later succeeds — which is |
| 28 | // exactly what a caller cannot work out from an error STRING, hence a |
| 29 | // sentinel to match with errors.Is. |
| 30 | var ErrBusy = errors.New("ledger: append contention") |
| 31 | |
| 32 | // Append validates, chains, and writes one entry, advancing HEAD with CAS. |
| 33 | // A lost HEAD race re-reads and re-chains (up to 4 attempts). An entry |
| 34 | // object orphaned by a lost race is harmless: nothing references it, and |
| 35 | // Verify walks only the chain. |
| 36 | func claimRawAppend(ctx context.Context, st release.Store, sqid, app, name, writer string, |
| 37 | payload json.RawMessage, now func() time.Time) (Appended, error) { |
| 38 | if !ValidName(name) { |
| 39 | return Appended{}, fmt.Errorf("ledger: invalid ledger name %q", name) |
| 40 | } |
| 41 | if len(payload) > MaxPayload { |
| 42 | return Appended{}, fmt.Errorf("ledger: payload %d bytes exceeds cap %d", len(payload), MaxPayload) |
| 43 | } |
| 44 | if !json.Valid(payload) { |
| 45 | return Appended{}, fmt.Errorf("ledger: payload is not valid JSON") |
| 46 | } |
| 47 | |
| 48 | var lastErr error |
| 49 | for range appendAttempts { |
| 50 | headKey := HeadKey(sqid, app, name) |
| 51 | hb, etag, err := st.Get(ctx, headKey) |
| 52 | var head Head |
| 53 | exists := err == nil |
| 54 | switch { |
| 55 | case errors.Is(err, os.ErrNotExist): |
| 56 | // First entry: prev fields stay empty, HEAD is create-only. |
| 57 | case err != nil: |
| 58 | return Appended{}, err |
| 59 | default: |
| 60 | if err := json.Unmarshal(hb, &head); err != nil { |
| 61 | return Appended{}, fmt.Errorf("ledger: %s corrupt: %w", headKey, err) |
| 62 | } |
| 63 | } |
| 64 | |
| 65 | ts := now().UTC().Format(time.RFC3339) |
| 66 | e := Entry{ |
| 67 | Schema: Schema, Ledger: name, Writer: writer, TS: ts, |
| 68 | Payload: payload, PrevKey: head.HeadKey, PrevSHA256: head.HeadSHA256, |
| 69 | } |
| 70 | eb, err := json.Marshal(e) |
| 71 | if err != nil { |
| 72 | return Appended{}, err |
| 73 | } |
| 74 | sha := SHA256Hex(eb) |
| 75 | key := EntryKey(sqid, app, name, ts, sha) |
| 76 | if _, err := st.Put(ctx, key, eb, ""); err != nil { |
| 77 | if errors.Is(err, release.ErrCASConflict) { |
| 78 | // Same second, same bytes-shape collision: retry re-reads |
| 79 | // HEAD and produces a differently-chained entry. |
| 80 | lastErr = err |
| 81 | continue |
| 82 | } |
| 83 | return Appended{}, err |
| 84 | } |
| 85 | |
| 86 | nh := Head{Schema: Schema, UpdatedAt: ts, HeadKey: key, HeadSHA256: sha, Count: head.Count + 1} |
| 87 | nhb, err := json.Marshal(nh) |
| 88 | if err != nil { |
| 89 | return Appended{}, err |
| 90 | } |
| 91 | ifMatch := "" |
| 92 | if exists { |
| 93 | ifMatch = etag |
| 94 | } |
| 95 | if _, err := st.Put(ctx, headKey, nhb, ifMatch); err != nil { |
| 96 | if errors.Is(err, release.ErrCASConflict) { |
| 97 | // Lost the race: another writer advanced HEAD. Our entry |
| 98 | // object is now an unreferenced orphan — harmless in a |
| 99 | // versioned bucket; Verify walks only the chain. Re-read |
| 100 | // and re-chain. |
| 101 | lastErr = err |
| 102 | continue |
| 103 | } |
| 104 | return Appended{}, err |
| 105 | } |
| 106 | return Appended{Key: key, SHA256: sha, Count: nh.Count}, nil |
| 107 | } |
| 108 | return Appended{}, fmt.Errorf("ledger: append lost %d CAS races: %w: %w", appendAttempts, ErrBusy, lastErr) |
| 109 | } |
| 110 | |
amadan