amadan

carlos / platform Public

Replace the quantity audit guard only when its body changed, in one transaction

5f2b4faf dropped and recreated the processing guard on every open so a
database created before the landed_unverified outcome would admit it.
Two console processes opening the same database could meet between the
drop and the create, and the second create would fail the open.

The guard's stored text is now compared with the current one. It is
replaced only when they differ, in one immediate transaction, and a
database that is already current is not touched.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude b4e272ffa92462bb09f93ea7908ceebb1a19055b parent daa32cb
2 files changed, +80 −17
  • internal/console/billingquantity_test.go +51 −0
  • internal/console/billingquantityaudit.go +29 −17
diff --git a/internal/console/billingquantity_test.go b/internal/console/billingquantity_test.go
index 55a3dea23..3289285ae 100644
--- a/internal/console/billingquantity_test.go
+++ b/internal/console/billingquantity_test.go
@@ -6,6 +6,7 @@ import (
"fmt"
"net/http"
"net/http/httptest"
+ "path/filepath"
"strconv"
"strings"
"sync"
@@ -1210,3 +1211,53 @@ func TestQuantityWorkerFinishesAnOutstandingOperation(t *testing.T) {
}
assertLastAddition(t, fx, quantityOutcomeCharged, "si_team:quantity=11:always_invoice")
}
+
+// The processing guard names the outcomes it admits. A database whose guard
+// predates an outcome gets the current guard on the next open; a database
+// that is already current is left alone; and opening it again, as a second
+// console process does, is not an error.
+func TestQuantityProcessingGuardIsReplacedWhenItsBodyChanges(t *testing.T) {
+ path := filepath.Join(t.TempDir(), "console.db")
+ st, err := Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ guardSQL := func(s *Store) string {
+ var body string
+ if err := s.db.QueryRow(`SELECT sql FROM sqlite_master WHERE type='trigger' AND name='billing_quantity_events_processing_guard'`).Scan(&body); err != nil {
+ t.Fatal(err)
+ }
+ return body
+ }
+ current := guardSQL(st)
+ if !strings.Contains(current, "'landed_unverified'") {
+ t.Fatalf("the guard does not admit landed_unverified: %s", current)
+ }
+ // An older build's guard, without the new outcome.
+ old := strings.Replace(current, "'landed_unverified',", "", 1)
+ if _, err := st.db.Exec(`DROP TRIGGER billing_quantity_events_processing_guard; ` + old); err != nil {
+ t.Fatal(err)
+ }
+ if strings.Contains(guardSQL(st), "'landed_unverified'") {
+ t.Fatal("fixture: the old guard still admits the new outcome")
+ }
+ // A second process opens the same database while the first is open.
+ second, err := Open(path)
+ if err != nil {
+ t.Fatalf("a second open of the same database: %v", err)
+ }
+ defer second.Close()
+ if got := guardSQL(st); got != current {
+ t.Fatalf("the stale guard was not replaced:\n%s", got)
+ }
+ // And a third open of a current database changes nothing and is fine.
+ third, err := Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer third.Close()
+ if got := guardSQL(third); got != current {
+ t.Fatal("a current guard was rewritten differently")
+ }
+ st.Close()
+}
diff --git a/internal/console/billingquantityaudit.go b/internal/console/billingquantityaudit.go
index 070bd34e9..34199aec2 100644
--- a/internal/console/billingquantityaudit.go
+++ b/internal/console/billingquantityaudit.go
@@ -1,6 +1,9 @@
package console
-import "database/sql"
+import (
+ "database/sql"
+ "errors"
+)
func (s *Store) migrateQuantityHistoryGuards() error {
for _, col := range []string{"processing_outcome", "processing_reference"} {
@@ -8,29 +11,38 @@ func (s *Store) migrateQuantityHistoryGuards() error {
return err
}
}
- for _, statement := range []string{
- `CREATE TRIGGER IF NOT EXISTS billing_quantity_events_insert_guard BEFORE INSERT ON billing_quantity_events
+ // The processing guard's body names the outcomes it admits, so a
+ // database created before an outcome was added keeps a guard that
+ // refuses it. The guard is replaced when its stored text differs from
+ // this one, and only then, in one immediate transaction: two console
+ // processes opening the same database cannot meet between the drop and
+ // the create, and a database that is already current is not touched.
+ const processingGuard = `CREATE TRIGGER billing_quantity_events_processing_guard BEFORE UPDATE ON billing_quantity_events
+ WHEN NEW.processed_at<=0 OR NEW.processed_at<OLD.occurred_at_ns/1000000000
+ OR NEW.processing_outcome NOT IN ('charged','included','landed_unverified','checkout_expired','checkout_abandoned')
+ OR length(trim(NEW.processing_reference))=0 OR length(NEW.processing_reference)>256
+ BEGIN SELECT RAISE(ABORT,'billing_quantity_processing_invalid'); END`
+ if _, err := s.db.Exec(`CREATE TRIGGER IF NOT EXISTS billing_quantity_events_insert_guard BEFORE INSERT ON billing_quantity_events
WHEN NEW.processed_at<>0 OR NEW.processing_outcome<>'' OR NEW.processing_reference<>''
OR NOT ((NEW.change='addition' AND NEW.effect='prorate' AND NEW.members_after=NEW.members_before+1)
OR (NEW.change='removal' AND NEW.effect='at_renewal' AND NEW.members_after=NEW.members_before-1))
OR EXISTS(SELECT 1 FROM billing_quantity_events WHERE seq=NEW.seq OR (account_id=NEW.account_id AND generation=NEW.generation))
- BEGIN SELECT RAISE(ABORT,'billing_quantity_events_append_only'); END`,
- // Dropped and recreated, not IF NOT EXISTS: the set of outcomes the
- // guard admits is part of its body, and a database created before
- // an outcome was added would otherwise keep refusing it.
- `DROP TRIGGER IF EXISTS billing_quantity_events_processing_guard`,
- `CREATE TRIGGER billing_quantity_events_processing_guard BEFORE UPDATE ON billing_quantity_events
- WHEN NEW.processed_at<=0 OR NEW.processed_at<OLD.occurred_at_ns/1000000000
- OR NEW.processing_outcome NOT IN ('charged','included','landed_unverified','checkout_expired','checkout_abandoned')
- OR length(trim(NEW.processing_reference))=0 OR length(NEW.processing_reference)>256
- BEGIN SELECT RAISE(ABORT,'billing_quantity_processing_invalid'); END`,
- `UPDATE billing_quantity_changes SET history_gap=1,state='history_unproven'
- WHERE EXISTS(SELECT 1 FROM billing_quantity_events e WHERE e.account_id=billing_quantity_changes.account_id AND e.processed_at<>0 AND (e.processed_at<e.occurred_at_ns/1000000000 OR e.processing_outcome NOT IN ('charged','included','landed_unverified','checkout_expired','checkout_abandoned') OR length(trim(e.processing_reference))=0 OR length(e.processing_reference)>256))`,
- } {
- if _, err := s.db.Exec(statement); err != nil {
+ BEGIN SELECT RAISE(ABORT,'billing_quantity_events_append_only'); END`); err != nil {
+ return err
+ }
+ var stored sql.NullString
+ if err := s.db.QueryRow(`SELECT sql FROM sqlite_master WHERE type='trigger' AND name='billing_quantity_events_processing_guard'`).Scan(&stored); err != nil && !errors.Is(err, sql.ErrNoRows) {
+ return err
+ }
+ if !stored.Valid || stored.String != processingGuard {
+ if _, err := s.db.Exec(`BEGIN IMMEDIATE; DROP TRIGGER IF EXISTS billing_quantity_events_processing_guard; ` + processingGuard + `; COMMIT`); err != nil {
return err
}
}
+ if _, err := s.db.Exec(`UPDATE billing_quantity_changes SET history_gap=1,state='history_unproven'
+ WHERE EXISTS(SELECT 1 FROM billing_quantity_events e WHERE e.account_id=billing_quantity_changes.account_id AND e.processed_at<>0 AND (e.processed_at<e.occurred_at_ns/1000000000 OR e.processing_outcome NOT IN ('charged','included','landed_unverified','checkout_expired','checkout_abandoned') OR length(trim(e.processing_reference))=0 OR length(e.processing_reference)>256))`); err != nil {
+ return err
+ }
return nil
}