DurableEmitter: LOOP Plugin Support #2073
+343
−61
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -224,7 +224,7 @@ func TestDurableEmitter_HooksPublishFailureSkipsMarkHook(t *testing.T) { | |
| assert.Equal(t, int32(0), markCalls.Load()) | ||
| } | ||
|
|
||
| func TestDurableEmitter_NoRetransmitSkipsRetransmitAndExpiry(t *testing.T) { | ||
| store := NewMemDurableEventStore() | ||
| client := &testChipClient{} | ||
| client.setPublishErr(errors.New("chip unavailable")) | ||
|
|
@@ -249,14 +249,14 @@ func TestDurableEmitter_NonHostProcessSkipsRetransmitAndExpiry(t *testing.T) { | |
| return client.batchCount.Load() >= 1 && store.Len() == 1 | ||
| }, 2*time.Second, 5*time.Millisecond, "initial PublishBatch should fail and leave the row") | ||
|
|
||
| // Several retransmit-only ticks would have cleared or retried by now. | ||
| time.Sleep(250 * time.Millisecond) | ||
|
|
||
| assert.Equal(t, 1, store.Len(), "retransmit=false must not run retransmit or expiry loops") | ||
| assert.Equal(t, int64(1), client.batchCount.Load(), "retransmit=false must not schedule extra PublishBatch via retransmit") | ||
| } | ||
|
|
||
| func TestDurableEmitter_NoRetransmitStillDeliversViaBatchWorkers(t *testing.T) { | ||
| store := NewMemDurableEventStore() | ||
| client := &testChipClient{} | ||
|
|
||
|
|
@@ -272,7 +272,7 @@ func TestDurableEmitter_NonHostProcessStillDeliversViaBatchWorkers(t *testing.T) | |
|
|
||
| require.Eventually(t, func() bool { | ||
| return store.Len() == 0 && client.batchCount.Load() >= 1 | ||
| }, 2*time.Second, 10*time.Millisecond, "batch publish workers must still run when retransmit is false") | ||
| } | ||
|
|
||
| func TestDurableEmitter_EmitPersistsAndPublishes(t *testing.T) { | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,209 @@ | ||
| // Package durable_events provides a Postgres-backed implementation of | ||
| // beholder.DurableEventStore. It is kept in a sibling package to pkg/beholder | ||
| // so that consumers of the beholder API (including builds targeting wasip1) | ||
| // do not transitively import lib/pq. | ||
| package durable_events | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "strings" | ||
| "time" | ||
|
|
||
| "github.com/lib/pq" | ||
|
|
||
| "github.com/smartcontractkit/chainlink-common/pkg/beholder" | ||
| "github.com/smartcontractkit/chainlink-common/pkg/sqlutil" | ||
| ) | ||
|
|
||
| const chipDurableEventsTable = "cre.chip_durable_events" | ||
|
|
||
| // Store is a Postgres-backed implementation of beholder.DurableEventStore. | ||
| type Store struct { | ||
| ds sqlutil.DataSource | ||
| } | ||
|
|
||
| var ( | ||
| _ beholder.DurableEventStore = (*Store)(nil) | ||
| _ beholder.DurableQueueObserver = (*Store)(nil) | ||
| _ beholder.BatchInserter = (*Store)(nil) | ||
| ) | ||
|
|
||
| // New returns a Postgres-backed DurableEventStore bound to ds. | ||
| func New(ds sqlutil.DataSource) *Store { | ||
| return &Store{ds: ds} | ||
| } | ||
|
|
||
| func (s *Store) Insert(ctx context.Context, payload []byte) (int64, error) { | ||
| const q = `INSERT INTO ` + chipDurableEventsTable + ` (payload) VALUES ($1) RETURNING id` | ||
| var id int64 | ||
| if err := s.ds.GetContext(ctx, &id, q, payload); err != nil { | ||
| return 0, fmt.Errorf("failed to insert chip durable event: %w", err) | ||
| } | ||
| return id, nil | ||
| } | ||
|
|
||
| func (s *Store) InsertBatch(ctx context.Context, payloads [][]byte) ([]int64, error) { | ||
| if len(payloads) == 0 { | ||
| return nil, nil | ||
| } | ||
| placeholders := make([]string, len(payloads)) | ||
| args := make([]interface{}, len(payloads)) | ||
| for i, p := range payloads { | ||
| placeholders[i] = fmt.Sprintf("($%d)", i+1) | ||
| args[i] = p | ||
| } | ||
| q := fmt.Sprintf( | ||
| "INSERT INTO %s (payload) VALUES %s RETURNING id", | ||
| chipDurableEventsTable, | ||
| strings.Join(placeholders, ","), | ||
| ) | ||
|
|
||
| var ids []int64 | ||
| if err := s.ds.SelectContext(ctx, &ids, q, args...); err != nil { | ||
| return nil, fmt.Errorf("failed to batch insert chip durable events: %w", err) | ||
| } | ||
| return ids, nil | ||
| } | ||
|
|
||
| func (s *Store) Delete(ctx context.Context, id int64) error { | ||
| const q = `DELETE FROM ` + chipDurableEventsTable + ` WHERE id = $1` | ||
| if _, err := s.ds.ExecContext(ctx, q, id); err != nil { | ||
| return fmt.Errorf("failed to delete chip durable event id=%d: %w", id, err) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (s *Store) MarkDelivered(ctx context.Context, id int64) error { | ||
| const q = `UPDATE ` + chipDurableEventsTable + ` SET delivered_at = now() WHERE id = $1 AND delivered_at IS NULL` | ||
| if _, err := s.ds.ExecContext(ctx, q, id); err != nil { | ||
| return fmt.Errorf("failed to mark chip durable event delivered id=%d: %w", id, err) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (s *Store) MarkDeliveredBatch(ctx context.Context, ids []int64) (int64, error) { | ||
| if len(ids) == 0 { | ||
| return 0, nil | ||
| } | ||
| const q = `UPDATE ` + chipDurableEventsTable + ` SET delivered_at = now() WHERE id = ANY($1) AND delivered_at IS NULL` | ||
| res, err := s.ds.ExecContext(ctx, q, pq.Array(ids)) | ||
| if err != nil { | ||
| return 0, fmt.Errorf("failed to batch mark chip durable events delivered: %w", err) | ||
| } | ||
| n, _ := res.RowsAffected() | ||
| return n, nil | ||
| } | ||
|
|
||
| func (s *Store) PurgeDelivered(ctx context.Context, batchLimit int) (int64, error) { | ||
| if batchLimit <= 0 { | ||
| return 0, nil | ||
| } | ||
| const q = ` | ||
| WITH picked AS ( | ||
| SELECT id FROM ` + chipDurableEventsTable + ` | ||
| WHERE delivered_at IS NOT NULL | ||
| ORDER BY delivered_at ASC | ||
| LIMIT $1 | ||
| ) | ||
| DELETE FROM ` + chipDurableEventsTable + ` AS t | ||
| USING picked WHERE t.id = picked.id` | ||
| res, err := s.ds.ExecContext(ctx, q, batchLimit) | ||
| if err != nil { | ||
| return 0, fmt.Errorf("failed to purge delivered chip durable events: %w", err) | ||
| } | ||
| n, err := res.RowsAffected() | ||
| if err != nil { | ||
| return 0, fmt.Errorf("purge delivered rows affected: %w", err) | ||
| } | ||
| return n, nil | ||
| } | ||
|
|
||
| func (s *Store) ListPending(ctx context.Context, createdBefore time.Time, limit int) ([]beholder.DurableEvent, error) { | ||
| const q = ` | ||
| SELECT id, payload, created_at | ||
| FROM ` + chipDurableEventsTable + ` | ||
| WHERE delivered_at IS NULL | ||
| AND created_at < $1 | ||
| ORDER BY created_at ASC | ||
| LIMIT $2` | ||
|
|
||
| type row struct { | ||
| ID int64 `db:"id"` | ||
| Payload []byte `db:"payload"` | ||
| CreatedAt time.Time `db:"created_at"` | ||
| } | ||
|
|
||
| var rows []row | ||
| if err := s.ds.SelectContext(ctx, &rows, q, createdBefore, limit); err != nil { | ||
| return nil, fmt.Errorf("failed to list pending chip durable events: %w", err) | ||
| } | ||
|
|
||
| out := make([]beholder.DurableEvent, 0, len(rows)) | ||
| for _, r := range rows { | ||
| out = append(out, beholder.DurableEvent{ | ||
| ID: r.ID, | ||
| Payload: r.Payload, | ||
| CreatedAt: r.CreatedAt, | ||
| }) | ||
| } | ||
| return out, nil | ||
| } | ||
|
|
||
| func (s *Store) DeleteExpired(ctx context.Context, ttl time.Duration) (int64, error) { | ||
| const q = ` | ||
| WITH deleted AS ( | ||
| DELETE FROM ` + chipDurableEventsTable + ` | ||
| WHERE created_at <= now() - $1::interval | ||
| RETURNING id | ||
| ) | ||
| SELECT count(*) FROM deleted` | ||
|
|
||
| var count int64 | ||
| if err := s.ds.GetContext(ctx, &count, q, ttl.String()); err != nil { | ||
| return 0, fmt.Errorf("failed to delete expired chip durable events: %w", err) | ||
| } | ||
| return count, nil | ||
| } | ||
|
|
||
| type chipDurableQueueAgg struct { | ||
| Cnt int64 `db:"cnt"` | ||
| PayloadSum int64 `db:"payload_sum"` | ||
| MinCreated *time.Time `db:"min_created"` | ||
| } | ||
|
|
||
| // ObserveDurableQueue implements beholder.DurableQueueObserver for queue depth / age gauges. | ||
| func (s *Store) ObserveDurableQueue(ctx context.Context, eventTTL, nearExpiryLead time.Duration) (beholder.DurableQueueStats, error) { | ||
| const qAgg = ` | ||
| SELECT | ||
| count(*)::bigint AS cnt, | ||
| coalesce(sum(octet_length(payload)), 0)::bigint AS payload_sum, | ||
| min(created_at) AS min_created | ||
| FROM ` + chipDurableEventsTable + ` | ||
| WHERE delivered_at IS NULL` | ||
|
|
||
| var row chipDurableQueueAgg | ||
| if err := s.ds.GetContext(ctx, &row, qAgg); err != nil { | ||
| return beholder.DurableQueueStats{}, fmt.Errorf("durable queue aggregate: %w", err) | ||
| } | ||
| var st beholder.DurableQueueStats | ||
| st.Depth = row.Cnt | ||
| st.PayloadBytes = row.PayloadSum | ||
| if row.MinCreated != nil { | ||
| st.OldestPendingAge = time.Since(*row.MinCreated) | ||
| } | ||
| if eventTTL > 0 && nearExpiryLead > 0 && nearExpiryLead < eventTTL { | ||
| ttlSec := int64(eventTTL.Round(time.Second) / time.Second) | ||
| leadSec := int64(nearExpiryLead.Round(time.Second) / time.Second) | ||
| const qNear = ` | ||
| SELECT count(*)::bigint | ||
| FROM ` + chipDurableEventsTable + ` | ||
| WHERE delivered_at IS NULL | ||
| AND created_at >= now() - ($1::bigint * interval '1 second') | ||
| AND created_at < now() - (($1::bigint - $2::bigint) * interval '1 second')` | ||
| if err := s.ds.GetContext(ctx, &st.NearTTLCount, qNear, ttlSec, leadSec); err != nil { | ||
| return beholder.DurableQueueStats{}, fmt.Errorf("durable queue near-ttl: %w", err) | ||
| } | ||
| } | ||
| return st, nil | ||
| } |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.