Class: Hecks::Adapters::PostgresEra

Inherits:
Object
  • Object
show all
Includes:
SqlQueryBuilder
Defined in:
lib/hecks/adapters/driven/postgres_era.rb,
lib/hecks/adapters/driven/postgres_era/lineage.rb,
lib/hecks/adapters/driven/postgres_era/lineage_manager.rb,
lib/hecks/adapters/driven/postgres_era/lineage/era_store.rb,
lib/hecks/adapters/driven/postgres_era/lineage/tail_merge.rb,
lib/hecks/adapters/driven/postgres_era/lineage/field_cache.rb,
lib/hecks/adapters/driven/postgres_era/lineage/provisioning.rb,
lib/hecks/adapters/driven/postgres_era/lineage/head_compiler.rb,
lib/hecks/adapters/driven/postgres_era/lineage_manager/minter.rb,
lib/hecks/adapters/driven/postgres_era/lineage/mint_transaction.rb,
lib/hecks/adapters/driven/postgres_era/lineage/resumable_backfill.rb,
lib/hecks/adapters/driven/postgres_era/lineage/transform_installer.rb,
lib/hecks/adapters/driven/postgres_era/lineage_manager/era_resolver.rb,
lib/hecks/adapters/driven/postgres_era/lineage_manager/coverage_check.rb,
lib/hecks/adapters/driven/postgres_era/lineage_manager/merge_coordinator.rb

Overview

The enforcement-grade persistence adapter — and the only one that declares the LINEAGE capability: it may act on shape drift (translate, fork, merge) where every other adapter can only refuse toward it. Sibling to the plain Postgres adapter (postgres.rb), which is the same database with none of this machinery — pick PostgresEra only once a domain actually needs to survive a shape change live. See docs/implemented/postgres-era-adapter-split-plan.md for why the two are split and what each one carries.

Storage model (see postgres_era/lineage.rb for the DDL):

  • One journal per DOMAIN, list-partitioned by era, one ordinal sequence spanning partitions. Appends go there; nothing updates or deletes a journal row (immutability by privilege — UPDATE and DELETE revoked; a deployment's app role connects as a non-owner).
  • Per aggregate, the HEAD is derived: era 1 reads a plain view (latest save per id); later eras read a view overlaying the materialized, translated ancestor tail with live current-era rows. project is therefore a no-op — old entries are never rewritten, and the head is never a table anything writes.
  • State is ONE jsonb column. jsonb normalizes key order (and drops duplicate keys), so anything comparing stored state — the corpus history gate above all — must compare CANONICALIZED state, never raw bytes; bin/canonicalise deep-sorts keys, which is exactly why the gate survives this normalization.
  • Query pushdown is the shared SqlQueryBuilder: every declared operator compiles fully into SQL, or the query refuses loudly.

Defined Under Namespace

Modules: LineageManager Classes: Lineage

Constant Summary

Constants included from SqlQueryBuilder

SqlQueryBuilder::COMPARATORS

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(aggregate:, settings: {}, root: nil) ⇒ PostgresEra

Returns a new instance of PostgresEra.



129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
 
# File 'lib/hecks/adapters/driven/postgres_era.rb', line 129

def initialize(aggregate:, settings: {}, root: nil)
  @aggregate = aggregate
  @db = self.class.connect_for(aggregate.name, settings)
  # The domain names the journal (one journal per lineage). The
  # factory injects it; a directly-instantiated adapter (specs,
  # consoles) journals under the aggregate's own name.
  @domain = (settings[:domain] || settings["domain"] || aggregate.storage_name).to_s
  @lineage = Lineage.new(@db, @domain)
  @lineage.ensure_base!
  # The era gate resolves which era this boot IS (an old checkout
  # boots a held-but-superseded era and keeps writing its own
  # partition); a directly-instantiated adapter defaults to the
  # newest.
  @era = settings[:era] || settings["era"] || @lineage.current_era
  # Unconditional and idempotent, regardless of era — belt-and-
  # suspenders self-healing (compile_head! already ensures this for
  # a freshly-minted era's own name; ensure_first_head! for era 1's)
  # against any boot-ordering surprise, at the cost of one
  # CREATE TABLE IF NOT EXISTS nobody pays for twice.
  @lineage.ensure_head_snapshot!(table, @era)
  @lineage.ensure_first_head!(table) if @era == 1
  # THE READ-CACHE SIDE OF THE ERA WORKAROUND (Track C,
  # docs/implemented/postgres-era-adapter-split-plan.md §3) — one row-cache
  # table per `where`-field this aggregate's own declared queries
  # (and its entities' own) actually use, derived automatically
  # (principle 3 — no bluebook keyword), self-healing and
  # idempotent like everything else booted here. `@field_caches`
  # maps field -> cache-table name; `query` below consults it to
  # decide whether a declared query can skip the DISTINCT ON
  # reduction entirely.
  @field_caches = ensure_field_caches!
  create_event_table!
  create_saga_table!
end

Instance Attribute Details

#aggregate ⇒ Object (readonly)

Returns the value of attribute aggregate.



44
45
46
 
# File 'lib/hecks/adapters/driven/postgres_era.rb', line 44

def aggregate
  @aggregate
end

Class Method Details

.connect_for(name, settings) ⇒ Object 



70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
 
# File 'lib/hecks/adapters/driven/postgres_era.rb', line 70

def self.connect_for(name, settings)
  # LAZY, ON PURPOSE — same reasoning as Sqlite's own initialize:
  # a domain that never wires PostgresEra should never need the gem.
  require "pg"

  declared = settings[:database] || settings["database"]
  if declared.to_s.empty?
    raise Runtime::WiringError,
          "#{name} binds PostgresEra, which needs a database connection, " \
          "but its world declares no \"database\"."
  end

  connection =
    if declared.start_with?("postgres://", "postgresql://")
      PG.connect(declared)
    else
      PG.connect(dbname: declared)
    end

  # SHARED-INSTANCE ISOLATION. A domain that declares `schema` is
  # sharing its Postgres instance with other domains (the
  # storehouse) — every unqualified table/view/function reference
  # this adapter and its lineage classes ever construct resolves
  # through search_path, so this one SET is what makes ALTER
  # TABLE ... SET SCHEMA migrations transparent to the rest of the
  # adapter. A domain with no `schema` setting keeps Postgres's
  # own default search_path (public), same as before this existed.
  schema = settings[:schema] || settings["schema"]
  if schema.to_s != ""
    # THE SCHEMA ITSELF, IDEMPOTENTLY — a domain naming a `schema:`
    # nobody has created yet used to fail on its FIRST table-
    # creation attempt with Postgres's own "no schema has been
    # selected to create in", found live provisioning tenant_
    # isolation_spec.rb's own multi-schema fixture by hand before
    # this existed. `CREATE SCHEMA IF NOT EXISTS` is exactly the
    # same self-healing idempotency this adapter's own table/era
    # provisioning already holds itself to (see the comment right
    # below on `client_min_messages`) — a schema that already
    # exists is the ORDINARY case for every boot after the first,
    # not news.
    connection.exec("CREATE SCHEMA IF NOT EXISTS #{connection.quote_ident(schema)}")
    connection.exec("SET search_path TO #{connection.quote_ident(schema)}")
  end

  # QUIET ON PURPOSE. Provisioning re-runs its own idempotent
  # `CREATE ... IF NOT EXISTS` checks on every boot — a schema that
  # already exists is the ORDINARY case, not news, and Postgres
  # surfaces every one as a NOTICE by default. `bin/set-password`
  # boots a real registry just to mint an Identity, and nobody
  # setting a password needs to see a page of "relation ...
  # already exists, skipping" to do it. WARNING and above (real
  # problems) still surface.
  connection.exec("SET client_min_messages = warning")
  connection
rescue PG::Error => error
  raise Runtime::WiringError,
        "cannot bind PostgresEra at #{declared} for #{name}: #{error.message.strip}"
end

.era_check!(registry:, bluebook:, current_text:, settings:, directory: nil) ⇒ Object 



63
64
65
66
67
68
 
# File 'lib/hecks/adapters/driven/postgres_era.rb', line 63

def self.era_check!(registry:, bluebook:, current_text:, settings:, directory: nil)
  LineageManager.check!(
    registry: registry, bluebook: bluebook, current_text: current_text,
    settings: settings, directory: directory
  )
end

.lineage_capable? ⇒ Boolean

The capability idiom: only PostgresEra answers true, and only PostgresEra carries an era_check! for the boot gate to delegate to.

Returns:

  • (Boolean) 


50
 
# File 'lib/hecks/adapters/driven/postgres_era.rb', line 50

def self.lineage_capable? = true

.tenant_capable? ⇒ Boolean

TENANT-CAPABLE — see Runtime::TenantCheck's own header for the full reasoning. connect_for's own schema: setting (the Storehouse shared-instance mechanism, already built, already proven for eras) is what keeps two tenant boots' tables apart: each boot's own SET search_path means every unqualified reference this adapter and its lineage classes construct resolves into that boot's own schema, never another tenant's — proven for real, not assumed, by tenant_isolation_spec.rb, the same discipline lineage_capable? already holds itself to.

Returns:

  • (Boolean) 


61
 
# File 'lib/hecks/adapters/driven/postgres_era.rb', line 61

def self.tenant_capable? = true

Instance Method Details

#all(order_by: nil, direction: :asc) ⇒ Object

order_by IS A RUNTIME VALUE, not framework-authored bluebook source like every other caller of order_expression — a query param off an HTTP request, in the console's case. Whitelisted against the aggregate's own real attributes (plus its lifecycle field) before it ever reaches order_expression, unlike a declared query's order_by, which the language itself already only lets name a real attribute at parse time. Without this, an unknown field wouldn't error — query_expression degrades a nil attribute to a harmless no-op path — it would just silently sort by nothing.

Raises:



182
183
184
185
186
187
188
189
190
 
# File 'lib/hecks/adapters/driven/postgres_era.rb', line 182

def all(order_by: nil, direction: :asc)
  return @db.exec(%(SELECT id, state FROM #{quoted_head} ORDER BY id)).map { |row| instance(row) } unless order_by

  name = order_by.to_s.split(".").first
  raise Runtime::WiringError, "#{@aggregate.name} has no attribute #{order_by.inspect} to order by" unless @aggregate.lifecycle&.field.to_s == name || @aggregate.attribute(name)

  spec = QuerySpecification::Common::OrderBy.new(field: order_by, direction: direction)
  @db.exec(%(SELECT id, state FROM #{quoted_head} ORDER BY #{order_clause(spec, nil)})).map { |row| instance(row) }
end

#append(entry) ⇒ Object

HELD FOR THE WHOLE TRANSACTION, not just around the INSERT — the ordinal is assigned by the column's own nextval() default, inside this same statement, so the lock has to already be held before that default evaluates. A DIFFERENT key from mint_era!/merge_tail!'s hecks_eras:domain : this serializes plain writes against EACH OTHER, never against a mint. See postgres/lineage.rb's own comment for why only that half of the race is closed. The journal insert and the snapshot upsert/delete happen in the SAME transaction — real ACID atomicity, not the append-then- project two-step a file-based adapter needs a crash-recovery replay for (see Heki). If this transaction commits, the snapshot is already exactly as current as the journal; if it doesn't, neither happened. project stays uninvolved on purpose — it still runs, cheaply, during AppendOnly#recover!'s full replay on every boot (see project below), and a second write there would make that replay pay real DB cost for a snapshot that's already correct.



255
256
257
258
259
260
261
 
# File 'lib/hecks/adapters/driven/postgres_era.rb', line 255

def append(entry)
  @db.transaction do
    lock_writes!
    append_and_project!(entry)
  end
  entry
end

#atomic_put(entry, insert_only: false) ⇒ Object

Outcome detection, journal append and every derived projection share the SAME transaction and domain write lock. The lineage-aware head determines whether this id is already visible; no repository find occurs before entering this adapter-native operation.



267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
 
# File 'lib/hecks/adapters/driven/postgres_era.rb', line 267

def atomic_put(entry, insert_only: false)
  status = nil
  @db.transaction do
    lock_writes!
    exists = !@db.exec_params(
      "SELECT 1 FROM #{quoted_head} WHERE id = $1 LIMIT 1",
      [entry.id.to_s]
    ).ntuples.zero?
    if insert_only && exists
      status = :conflicted
      next
    end
    status = exists ? :replaced : :inserted
    append_and_project!(entry)
  end
  status
end

#count ⇒ Object 



192
 
# File 'lib/hecks/adapters/driven/postgres_era.rb', line 192

def count = @db.exec(%(SELECT COUNT(*) FROM #{quoted_head}))[0]["count"].to_i

#delete(id) ⇒ Object 



323
324
325
326
327
 
# File 'lib/hecks/adapters/driven/postgres_era.rb', line 323

def delete(id)
  entry = Ports::Persistence::Entry.new(operation: "delete", id: id.to_s, state: nil)
  append(entry)
  true
end

#delete_saga(process_manager:, correlation:) ⇒ Object 



368
369
370
371
372
373
 
# File 'lib/hecks/adapters/driven/postgres_era.rb', line 368

def delete_saga(process_manager:, correlation:)
  @db.exec_params(
    "DELETE FROM hecks_saga_instances WHERE domain = $1 AND process_manager = $2 AND correlation = $3",
    [@domain, process_manager.to_s, correlation.to_s]
  )
end

#each_saga ⇒ Object 



375
376
377
378
379
380
381
382
383
384
385
 
# File 'lib/hecks/adapters/driven/postgres_era.rb', line 375

def each_saga
  return enum_for(:each_saga) unless block_given?

  @db.exec_params(
    "SELECT process_manager, correlation, state, memory FROM hecks_saga_instances WHERE domain = $1",
    [@domain]
  ).each do |row|
    yield row["process_manager"], row["correlation"], row["state"],
          JSON.parse(row["memory"], symbolize_names: true)
  end
end

#entries ⇒ Object 



296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
 
# File 'lib/hecks/adapters/driven/postgres_era.rb', line 296

def entries
  @db.exec_params(
    "SELECT aggregate_id, operation, state, mirrors FROM #{@lineage.quoted_journal} " \
    "WHERE aggregate = $1 ORDER BY ordinal",
    [table]
  ).map do |row|
    state = row["state"] && JSON.parse(row["state"])
    Ports::Persistence::Entry.new(
      operation: row["operation"] || "save",
      id:        row["aggregate_id"],
      state:     state&.transform_keys(&:to_sym),
      mirrors:   row["mirrors"] && JSON.parse(row["mirrors"])
    )
  end
end

#events ⇒ Object 



336
337
338
339
340
341
342
343
344
345
346
 
# File 'lib/hecks/adapters/driven/postgres_era.rb', line 336

def events
  @db.exec("SELECT * FROM events ORDER BY id").map do |row|
    Runtime::Event.new(
      name:        row["name"],
      aggregate:   row["aggregate"],
      id:          row["aggregate_id"],
      payload:     JSON.parse(row["payload"], symbolize_names: true),
      occurred_at: row["occurred_at"]
    )
  end
end

#find(id) ⇒ Object 



166
167
168
169
170
171
 
# File 'lib/hecks/adapters/driven/postgres_era.rb', line 166

def find(id)
  result = @db.exec_params(%(SELECT id, state FROM #{quoted_head} WHERE id = $1), [id.to_s])
  return nil if result.ntuples.zero?

  instance(result[0])
end

#persistence_capabilities ⇒ Object 



46
 
# File 'lib/hecks/adapters/driven/postgres_era.rb', line 46

def persistence_capabilities = [:atomic_put]

#project(entry) ⇒ Object

The head is DERIVED — projecting is reading, so there is nothing to write here. append above already keeps the snapshot the head view reads from current, transactionally. The instance is still built (and validated) so a save returns what every other adapter returns.



290
291
292
293
294
 
# File 'lib/hecks/adapters/driven/postgres_era.rb', line 290

def project(entry)
  return if entry.delete?

  Runtime::Instance.new(aggregate: @aggregate, id: entry.id, state: entry.state)
end

#query(declared, args = {}, context: {}) ⇒ Object

THE TWO-PHASE SHORTCUT (Track C, docs/implemented/postgres-era-adapter- split-plan.md §3). SqlQueryBuilder#query (super, unmodified per principle 2) always runs correctly here — it filters against head_view, which is already the fully-reduced current state — but for a domain that has minted a second era, that reduction itself is the expensive part, and no index on the jsonb path changes that (see field_cache.rb's own header for the SQL- semantics reason why). When every where clause this query declares either targets a cached field or is eligible to (an ordinary comparator, not a null-vs-value special case — see cache_eligible?), skip the reduction: look candidate ids up in the cache table(s) first (cheap, indexed, no reduction involved), then read ONLY those ids' current state from head_view — safe THROUGH the reduction because id is its own partition key. Any clause that ISN'T cache-eligible (an uncached field, or a null comparison) is simply re-checked against head_view in the second phase, exactly as super would have checked it anyway — this can only ever NARROW what phase two has to look at, never change what a clause means.

FALLS BACK TO super WHENEVER NO CLAUSE CAN BE ACCELERATED — a query with no where at all (order_by-only — no cache table exists for these, see field_cache.rb), a query whose only clauses target fields with no cache table, or a domain that has never minted a second era at all (@field_caches is never empty just because era 1 has no reduction to skip — the cache tables still exist and still accelerate era 1 the same way, but the fallback path is already just as cheap there since head_view IS the snapshot table verbatim for era 1; skipping straight to super in that case would be a valid FUTURE optimization, not attempted here to keep this one code path correct for every era uniformly).



225
226
227
228
229
230
231
232
233
234
235
236
 
# File 'lib/hecks/adapters/driven/postgres_era.rb', line 225

def query(declared, args = {}, context: {})
  return super if @field_caches.empty? || declared.wheres.empty?

  evaluated = declared.wheres.map { |clause| [clause, query_value(clause.value, args)] }
  cached, uncached = evaluated.partition { |clause, value| cache_eligible?(clause, value) }
  return super if cached.empty?

  ids = cache_phase(cached)
  return [] if ids.empty?

  head_phase(declared, uncached, ids, args)
end

#record_event(event) ⇒ Object 



329
330
331
332
333
334
 
# File 'lib/hecks/adapters/driven/postgres_era.rb', line 329

def record_event(event)
  @db.exec_params(
    "INSERT INTO events (name, aggregate, aggregate_id, payload, occurred_at) VALUES ($1, $2, $3, $4, $5)",
    [event.name, event.aggregate, event.id.to_s, JSON.generate(event.payload), event.occurred_at]
  )
end

#reset! ⇒ Object 



312
313
314
315
 
# File 'lib/hecks/adapters/driven/postgres_era.rb', line 312

def reset!
  @db.exec_params("DELETE FROM #{@lineage.quoted_journal} WHERE aggregate = $1", [table])
  self
end

#save(instance) ⇒ Object 



317
318
319
320
321
 
# File 'lib/hecks/adapters/driven/postgres_era.rb', line 317

def save(instance)
  entry = Ports::Persistence::Entry.new(operation: "save", id: instance.id.to_s, state: instance.state.dup)
  append(entry)
  project(entry)
end

#save_saga(process_manager:, correlation:, state:, memory:) ⇒ Object

── the OPTIONAL saga-persistence capability (Ports::Persistence's own three-method shape, §2) — one row per (domain, process_manager, correlation), domain kept as an explicit column even under schema isolation so two domains sharing one schema (neither declares its own schema) still isolate correctly, matching hecks_eras' own precedent (postgres/lineage/provisioning.rb). No advisory lock of its own: every call here already runs inside SagaInterpreter's own mutex (§7) serializing IN-PROCESS writers, and gets the SAME cross-process safety an aggregate's own writes get from this adapter — no better, no worse.



358
359
360
361
362
363
364
365
366
 
# File 'lib/hecks/adapters/driven/postgres_era.rb', line 358

def save_saga(process_manager:, correlation:, state:, memory:)
  @db.exec_params(
    "INSERT INTO hecks_saga_instances (domain, process_manager, correlation, state, memory) " \
    "VALUES ($1, $2, $3, $4, $5) " \
    "ON CONFLICT (domain, process_manager, correlation) DO UPDATE " \
    "SET state = EXCLUDED.state, memory = EXCLUDED.memory, updated_at = now()",
    [@domain, process_manager.to_s, correlation.to_s, state.to_s, JSON.generate(memory)]
  )
end

#table ⇒ Object 



164
 
# File 'lib/hecks/adapters/driven/postgres_era.rb', line 164

def table = @aggregate.storage_name