| 1 | package hostbox |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "database/sql" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "time" |
| 9 | ) |
| 10 | |
| 11 | // ExecuteServingPark records uncertainty before invoking filesystem authority. |
| 12 | // The callback runs without a SQLite transaction: it reads the same journal |
| 13 | // under root's namespace and host locks. A pending replay revalidates authority |
| 14 | // and resumes the deterministic retirement; it never rolls filesystem work back. |
| 15 | func (s *JournalService) ExecuteServingPark(ctx context.Context, peer string, q JournalRequest, run func(context.Context) (ServingParkResult, error)) (JournalResponse, error) { |
| 16 | out := JournalResponse{V: 1, ID: q.ID} |
| 17 | if err := validateServingParkRequest(q); err != nil { |
| 18 | return out, err |
| 19 | } |
| 20 | if s == nil || s.Launches == nil || s.Launches.db == nil || s.parkGate == nil || peer == "" || run == nil { |
| 21 | return out, ErrUnsupported |
| 22 | } |
| 23 | select { |
| 24 | case s.parkGate <- struct{}{}: |
| 25 | defer func() { <-s.parkGate }() |
| 26 | case <-ctx.Done(): |
| 27 | return out, ctx.Err() |
| 28 | } |
| 29 | pending, replayed, err := s.prepareServingPark(ctx, peer, q) |
| 30 | if err != nil || replayed { |
| 31 | return pending, err |
| 32 | } |
| 33 | result, err := run(ctx) |
| 34 | if err != nil { |
| 35 | // Even an identity refusal may follow a prior rename or a lost save |
| 36 | // reply. Preserve the pending receipt and the exact request for replay. |
| 37 | return out, errors.Join(ErrJournalUncertain, err) |
| 38 | } |
| 39 | out.ServingPark = &result |
| 40 | body, err := json.Marshal(out) |
| 41 | if err != nil { |
| 42 | return out, errors.Join(ErrJournalUncertain, err) |
| 43 | } |
| 44 | changed, err := s.Launches.db.ExecContext(ctx, "UPDATE journal_requests SET response=? WHERE id=? AND peer=? AND digest=?", body, q.ID, peer, journalRequestDigest(q)) |
| 45 | if err != nil { |
| 46 | return out, errors.Join(ErrJournalUncertain, err) |
| 47 | } |
| 48 | n, err := changed.RowsAffected() |
| 49 | if err != nil || n != 1 { |
| 50 | return out, ErrJournalUncertain |
| 51 | } |
| 52 | return out, nil |
| 53 | } |
| 54 | |
| 55 | func (s *JournalService) prepareServingPark(ctx context.Context, peer string, q JournalRequest) (JournalResponse, bool, error) { |
| 56 | out := JournalResponse{V: 1, ID: q.ID} |
| 57 | tx, err := s.Launches.db.BeginTx(ctx, nil) |
| 58 | if err != nil { |
| 59 | return out, false, err |
| 60 | } |
| 61 | defer tx.Rollback() |
| 62 | var oldPeer, digest string |
| 63 | var body []byte |
| 64 | err = tx.QueryRowContext(ctx, "SELECT peer,digest,response FROM journal_requests WHERE id=?", q.ID).Scan(&oldPeer, &digest, &body) |
| 65 | found := err == nil |
| 66 | if found { |
| 67 | if oldPeer != peer || digest != journalRequestDigest(q) { |
| 68 | return out, false, ErrChanged |
| 69 | } |
| 70 | if json.Unmarshal(body, &out) != nil || out.V != 1 || out.ID != q.ID { |
| 71 | return out, false, ErrJournalUncertain |
| 72 | } |
| 73 | if out.Error == "" && out.ServingPark != nil { |
| 74 | return out, true, nil |
| 75 | } |
| 76 | if out.Error != "uncertain" || out.ServingPark != nil { |
| 77 | return out, false, ErrJournalUncertain |
| 78 | } |
| 79 | } else if !errors.Is(err, sql.ErrNoRows) { |
| 80 | return out, false, err |
| 81 | } |
| 82 | now := time.Now() |
| 83 | if s.Now != nil { |
| 84 | now = s.Now() |
| 85 | } |
| 86 | var rejectBefore int64 |
| 87 | if err = tx.QueryRowContext(ctx, "SELECT reject_before FROM journal_receipt_retention WHERE id=1").Scan(&rejectBefore); err != nil { |
| 88 | return out, false, err |
| 89 | } |
| 90 | age := now.Sub(time.Unix(0, q.IssuedAt)) |
| 91 | if q.IssuedAt <= rejectBefore || age > JournalRequestWindow || age < -JournalRequestWindow { |
| 92 | return out, false, ErrJournalUncertain |
| 93 | } |
| 94 | if _, err = journalPrune(ctx, tx, rejectBefore, now); err != nil { |
| 95 | return out, false, err |
| 96 | } |
| 97 | if !found { |
| 98 | out.Error = "uncertain" |
| 99 | body, err = json.Marshal(out) |
| 100 | if err != nil { |
| 101 | return out, false, err |
| 102 | } |
| 103 | if _, err = tx.ExecContext(ctx, "INSERT INTO journal_requests(id,peer,digest,response,created,issued) VALUES(?,?,?,?,?,?)", q.ID, peer, journalRequestDigest(q), body, now.UnixNano(), q.IssuedAt); err != nil { |
| 104 | return out, false, err |
| 105 | } |
| 106 | } |
| 107 | return out, false, tx.Commit() |
| 108 | } |
| 109 | |
amadan