-
Notifications
You must be signed in to change notification settings - Fork 1.1k
Expand file tree
/
Copy pathmod.rs
More file actions
1827 lines (1620 loc) · 68.5 KB
/
mod.rs
File metadata and controls
1827 lines (1620 loc) · 68.5 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
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
128
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
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
mod state;
mod trigger_runner;
use crate::subgraph::context::IndexingContext;
use crate::subgraph::error::{
ClassifyErrorHelper as _, DetailHelper as _, NonDeterministicErrorHelper as _, ProcessingError,
ProcessingErrorKind,
};
use crate::subgraph::inputs::IndexingInputs;
use crate::subgraph::state::IndexingState;
use crate::subgraph::stream::new_block_stream;
use anyhow::Context as _;
use graph::blockchain::block_stream::{
BlockStream, BlockStreamEvent, BlockWithTriggers, FirehoseCursor,
};
use graph::blockchain::{
Block, BlockTime, Blockchain, DataSource as _, SubgraphFilter, Trigger, TriggerFilter as _,
TriggerFilterWrapper,
};
use graph::components::store::{EmptyStore, GetScope, ReadStore, StoredDynamicDataSource};
use graph::components::subgraph::InstanceDSTemplate;
use graph::components::trigger_processor::RunnableTriggers;
use graph::components::{
store::ModificationsAndCache,
subgraph::{MappingError, PoICausalityRegion, ProofOfIndexing, SharedProofOfIndexing},
};
use graph::data::store::scalar::Bytes;
use graph::data::subgraph::schema::SubgraphError;
use graph::data_source::{
offchain, CausalityRegion, DataSource, DataSourceCreationError, TriggerData,
};
use graph::env::EnvVars;
use graph::ext::futures::Cancelable;
use graph::futures03::stream::StreamExt;
use graph::prelude::{
anyhow, hex, retry, thiserror, BlockNumber, BlockPtr, BlockState, CancelGuard, CancelHandle,
CancelToken as _, CheapClone as _, EntityCache, EntityModification, Error,
InstanceDSTemplateInfo, LogCode, RunnerMetrics, RuntimeHostBuilder, StopwatchMetrics,
StoreError, StreamExtension, UnfailOutcome, Value, ENV_VARS,
};
use graph::schema::EntityKey;
use graph::slog::{debug, error, info, o, trace, warn, Logger};
use graph::util::lfu_cache::EvictStats;
use graph::util::{backoff::ExponentialBackoff, lfu_cache::LfuCache};
use std::sync::Arc;
use std::time::{Duration, Instant};
use std::vec;
use self::state::{RestartReason, RunnerState, StopReason};
use self::trigger_runner::TriggerRunner;
const MINUTE: Duration = Duration::from_secs(60);
const SKIP_PTR_UPDATES_THRESHOLD: Duration = Duration::from_secs(60 * 5);
const HANDLE_REVERT_SECTION_NAME: &str = "handle_revert";
const PROCESS_BLOCK_SECTION_NAME: &str = "process_block";
const PROCESS_TRIGGERS_SECTION_NAME: &str = "process_triggers";
const HANDLE_CREATED_DS_SECTION_NAME: &str = "handle_new_data_sources";
pub struct SubgraphRunner<C, T>
where
C: Blockchain,
T: RuntimeHostBuilder<C>,
{
ctx: IndexingContext<C, T>,
state: IndexingState,
inputs: Arc<IndexingInputs<C>>,
logger: Logger,
pub metrics: RunnerMetrics,
cancel_handle: Option<CancelHandle>,
/// The current state in the runner's state machine.
/// This field drives the main loop of the runner.
runner_state: RunnerState<C>,
}
#[derive(Debug, thiserror::Error)]
pub enum SubgraphRunnerError {
#[error("subgraph runner terminated because a newer one was active")]
Duplicate,
#[error(transparent)]
Unknown(#[from] Error),
}
impl<C, T> SubgraphRunner<C, T>
where
C: Blockchain,
T: RuntimeHostBuilder<C>,
{
pub fn new(
inputs: IndexingInputs<C>,
ctx: IndexingContext<C, T>,
logger: Logger,
metrics: RunnerMetrics,
env_vars: Arc<EnvVars>,
) -> Self {
Self {
inputs: Arc::new(inputs),
ctx,
state: IndexingState {
should_try_unfail_non_deterministic: true,
skip_ptr_updates_timer: Instant::now(),
backoff: ExponentialBackoff::with_jitter(
(MINUTE * 2).min(env_vars.subgraph_error_retry_ceil),
env_vars.subgraph_error_retry_ceil,
env_vars.subgraph_error_retry_jitter,
),
entity_lfu_cache: LfuCache::new(),
cached_head_ptr: None,
postponed_indexes_created: false,
},
logger,
metrics,
cancel_handle: None,
runner_state: RunnerState::Initializing,
}
}
/// Revert the state to a previous block. When handling revert operations
/// or failed block processing, it is necessary to remove part of the existing
/// in-memory state to keep it constent with DB changes.
/// During block processing new dynamic data sources are added directly to the
/// IndexingContext of the runner. This means that if, for whatever reason,
/// the changes don;t complete then the remnants of that block processing must
/// be removed. The same thing also applies to the block cache.
/// This function must be called before continuing to process in order to avoid
/// duplicated host insertion and POI issues with dirty entity changes.
fn revert_state_to(&mut self, block_number: BlockNumber) -> Result<(), Error> {
self.state.entity_lfu_cache = LfuCache::new();
// 1. Revert all hosts(created by DDS) at a block higher than `block_number`.
// 2. Unmark any offchain data sources that were marked done on the blocks being removed.
// When no offchain datasources are present, 2. should be a noop.
self.ctx.revert_data_sources(block_number + 1)?;
Ok(())
}
#[cfg(debug_assertions)]
pub fn context(&self) -> &IndexingContext<C, T> {
&self.ctx
}
#[cfg(debug_assertions)]
pub async fn run_for_test(self, break_on_restart: bool) -> Result<Self, Error> {
self.run_inner(break_on_restart).await.map_err(Into::into)
}
fn is_static_filters_enabled(&self) -> bool {
self.inputs.static_filters || self.ctx.hosts_len() > ENV_VARS.static_filters_threshold
}
fn build_filter(&self) -> TriggerFilterWrapper<C> {
let current_ptr = self.inputs.store.block_ptr();
let static_filters = self.is_static_filters_enabled();
// Filter out data sources that have reached their end block
let end_block_filter = |ds: &&C::DataSource| match current_ptr.as_ref() {
// We filter out datasources for which the current block is at or past their end block.
Some(block) => ds.end_block().is_none_or(|end| block.number < end),
// If there is no current block, we keep all datasources.
None => true,
};
let data_sources = self.ctx.static_data_sources();
let subgraph_filter = data_sources
.iter()
.filter_map(|ds| ds.as_subgraph())
.map(|ds| SubgraphFilter {
subgraph: ds.source.address(),
start_block: ds.source.start_block,
entities: ds
.mapping
.handlers
.iter()
.map(|handler| handler.entity.clone())
.collect(),
manifest_idx: ds.manifest_idx,
})
.collect::<Vec<_>>();
// if static_filters is not enabled we just stick to the filter based on all the data sources.
if !static_filters {
return TriggerFilterWrapper::new(
C::TriggerFilter::from_data_sources(
self.ctx.onchain_data_sources().filter(end_block_filter),
),
subgraph_filter,
);
}
// if static_filters is enabled, build a minimal filter with the static data sources and
// add the necessary filters based on templates.
// This specifically removes dynamic data sources based filters because these can be derived
// from templates AND this reduces the cost of egress traffic by making the payloads smaller.
if !self.inputs.static_filters {
info!(self.logger, "forcing subgraph to use static filters.")
}
let data_sources = self.ctx.static_data_sources();
let mut filter = C::TriggerFilter::from_data_sources(
data_sources
.iter()
.filter_map(|ds| ds.as_onchain())
// Filter out data sources that have reached their end block if the block is final.
.filter(end_block_filter),
);
let templates = self.ctx.templates();
filter.extend_with_template(templates.iter().filter_map(|ds| ds.as_onchain()).cloned());
TriggerFilterWrapper::new(filter, subgraph_filter)
}
#[cfg(debug_assertions)]
pub fn build_filter_for_test(&self) -> TriggerFilterWrapper<C> {
self.build_filter()
}
async fn start_block_stream(&mut self) -> Result<Cancelable<Box<dyn BlockStream<C>>>, Error> {
let block_stream_canceler = CancelGuard::new();
let block_stream_cancel_handle = block_stream_canceler.handle();
// TriggerFilter needs to be rebuilt eveytime the blockstream is restarted
let filter = self.build_filter();
let block_stream = new_block_stream(&self.inputs, filter, &self.metrics.subgraph)
.await?
.cancelable(&block_stream_canceler);
self.cancel_handle = Some(block_stream_cancel_handle);
// Keep the stream's cancel guard around to be able to shut it down when the subgraph
// deployment is unassigned
self.ctx
.instances
.insert(self.inputs.deployment.id, block_stream_canceler);
Ok(block_stream)
}
fn is_canceled(&self) -> bool {
if let Some(ref cancel_handle) = self.cancel_handle {
cancel_handle.is_canceled()
} else {
false
}
}
/// Initialize the runner by performing pre-loop setup.
///
/// This method handles:
/// - Updating the deployment synced metric
/// - Attempting to unfail deterministic errors from the previous run
/// - Checking if the subgraph has already reached its max end block
///
/// Returns the next state to transition to:
/// - `Restarting` to start the block stream (normal case)
/// - `Stopped` if the max end block was already reached
async fn initialize(&self) -> Result<RunnerState<C>, SubgraphRunnerError> {
self.update_deployment_synced_metric();
// If a subgraph failed for deterministic reasons, before start indexing, we first
// revert the deployment head. It should lead to the same result since the error was
// deterministic.
if let Some(current_ptr) = self.inputs.store.block_ptr() {
if let Some(parent_ptr) = self
.inputs
.triggers_adapter
.parent_ptr(¤t_ptr)
.await?
{
// This reverts the deployment head to the parent_ptr if
// deterministic errors happened.
//
// There's no point in calling it if we have no current or parent block
// pointers, because there would be: no block to revert to or to search
// errors from (first execution).
//
// We attempt to unfail deterministic errors to mitigate deterministic
// errors caused by wrong data being consumed from the providers. It has
// been a frequent case in the past so this helps recover on a larger scale.
let _outcome = self
.inputs
.store
.unfail_deterministic_error(¤t_ptr, &parent_ptr)
.await?;
}
// Stop subgraph when we reach maximum endblock.
if let Some(max_end_block) = self.inputs.max_end_block {
if max_end_block <= current_ptr.block_number() {
info!(self.logger, "Stopping subgraph as we reached maximum endBlock";
"max_end_block" => max_end_block,
"current_block" => current_ptr.block_number());
self.inputs.store.flush().await?;
return Ok(RunnerState::Stopped {
reason: StopReason::MaxEndBlockReached,
});
}
}
}
// Normal case: proceed to start the block stream
Ok(RunnerState::Restarting {
reason: RestartReason::StoreError, // Initial start uses the same path as restart
})
}
/// Await the next block stream event and transition to the appropriate state.
///
/// This method waits for the next event from the block stream and determines
/// which state the runner should transition to:
/// - `ProcessingBlock` for new blocks to process
/// - `Reverting` for revert events
/// - `Stopped` when the stream ends or is canceled
/// - Returns back to `AwaitingBlock` for non-fatal errors that allow continuation
async fn await_block(
&mut self,
mut block_stream: Cancelable<Box<dyn BlockStream<C>>>,
) -> Result<RunnerState<C>, SubgraphRunnerError> {
let event = {
let _section = self.metrics.stream.stopwatch.start_section("scan_blocks");
block_stream.next().await
};
if self.is_canceled() {
return self.cancel();
}
match event {
Some(Ok(BlockStreamEvent::ProcessBlock(block, cursor))) => {
Ok(RunnerState::ProcessingBlock {
block_stream,
block,
cursor,
})
}
Some(Ok(BlockStreamEvent::Revert(to_ptr, cursor))) => Ok(RunnerState::Reverting {
block_stream,
to_ptr,
cursor,
}),
// Log and drop the errors from the block_stream
// The block stream will continue attempting to produce blocks
Some(Err(e)) => {
// Log error and continue waiting for blocks
debug!(
&self.logger,
"Block stream produced a non-fatal error";
"error" => format!("{}", e),
);
Ok(RunnerState::AwaitingBlock { block_stream })
}
// If the block stream ends, that means that there is no more indexing to do.
None => Ok(RunnerState::Stopped {
reason: StopReason::StreamEnded,
}),
}
}
fn cancel(&mut self) -> Result<RunnerState<C>, SubgraphRunnerError> {
if self.ctx.instances.contains(&self.inputs.deployment.id) {
warn!(
self.logger,
"Terminating the subgraph runner because a newer one is active. \
Possible reassignment detected while the runner was in a non-cancellable pending state",
);
return Err(SubgraphRunnerError::Duplicate);
}
warn!(
self.logger,
"Terminating the subgraph runner because subgraph was unassigned",
);
Ok(RunnerState::Stopped {
reason: StopReason::Unassigned,
})
}
/// Construct a SubgraphError and mark the subgraph as failed in the store.
async fn fail_subgraph(
&mut self,
message: String,
block_ptr: Option<BlockPtr>,
deterministic: bool,
) -> Result<(), SubgraphRunnerError> {
let error = SubgraphError {
subgraph_id: self.inputs.deployment.hash.clone(),
message,
block_ptr,
handler: None,
deterministic,
};
self.inputs
.store
.fail_subgraph(error)
.await
.context("Failed to set subgraph status to `failed`")?;
Ok(())
}
/// Handle a restart by potentially restarting the store and starting a new block stream.
///
/// This method handles:
/// - Restarting the store if there were errors (to clear error state)
/// - Reverting state to the last good block if the store was restarted
/// - Starting a new block stream with updated filters
///
/// Returns the next state to transition to:
/// - `AwaitingBlock` with the new block stream (normal case)
async fn restart(
&mut self,
reason: RestartReason,
) -> Result<RunnerState<C>, SubgraphRunnerError> {
debug!(self.logger, "Starting or restarting subgraph"; "reason" => ?reason);
// If restarting due to store error, try to restart the store
if matches!(reason, RestartReason::StoreError) {
let store = self.inputs.store.cheap_clone();
if let Some(store) = store.restart().await? {
let last_good_block = store.block_ptr().map(|ptr| ptr.number).unwrap_or(0);
self.revert_state_to(last_good_block)?;
self.inputs = Arc::new(self.inputs.with_store(store));
}
}
let block_stream = self.start_block_stream().await?;
debug!(self.logger, "Started block stream");
self.metrics.subgraph.deployment_status.running();
self.update_deployment_synced_metric();
Ok(RunnerState::AwaitingBlock { block_stream })
}
/// Finalize the runner when it reaches a terminal state.
///
/// This method handles cleanup tasks when the runner stops:
/// - Flushing the store to ensure all changes are persisted
/// - Logging the stop reason
async fn finalize(self, reason: StopReason) -> Result<Self, SubgraphRunnerError> {
match reason {
StopReason::MaxEndBlockReached => {
info!(self.logger, "Stopping subgraph - max end block reached");
}
StopReason::Canceled => {
info!(self.logger, "Stopping subgraph - canceled");
}
StopReason::Unassigned => {
info!(self.logger, "Stopping subgraph - unassigned");
}
StopReason::StreamEnded => {
info!(self.logger, "Stopping subgraph - stream ended");
}
}
self.inputs.store.flush().await?;
Ok(self)
}
pub async fn run(self) -> Result<(), SubgraphRunnerError> {
self.run_inner(false).await.map(|_| ())
}
/// Main state machine loop for the subgraph runner.
///
/// This method drives the runner through its state machine, transitioning
/// between states based on events and actions. The state machine replaces
/// the previous nested loop structure with explicit state transitions.
///
/// ## State Machine
///
/// The runner starts in `Initializing` and transitions through states:
/// - `Initializing` → `Restarting` (or `Stopped` if max end block reached)
/// - `Restarting` → `AwaitingBlock`
/// - `AwaitingBlock` → `ProcessingBlock`, `Reverting`, or `Stopped`
/// - `ProcessingBlock` → `AwaitingBlock` or `Restarting`
/// - `Reverting` → `AwaitingBlock` or `Restarting`
/// - `Stopped` → terminal (returns)
async fn run_inner(mut self, break_on_restart: bool) -> Result<Self, SubgraphRunnerError> {
// Start in Initializing state
self.runner_state = RunnerState::Initializing;
// Track whether we've started processing blocks (not just initialized).
// This is used for break_on_restart logic - we should only stop on restart
// after we've actually started processing, not on the initial "restart"
// which is really the first start of the block stream.
let mut has_processed_blocks = false;
loop {
self.runner_state = match std::mem::take(&mut self.runner_state) {
RunnerState::Initializing => self.initialize().await?,
RunnerState::Restarting { reason } => {
if break_on_restart && has_processed_blocks {
// In test mode, stop on restart after first block processing
info!(self.logger, "Stopping subgraph on break");
RunnerState::Stopped {
reason: StopReason::Canceled,
}
} else {
self.restart(reason).await?
}
}
RunnerState::AwaitingBlock { block_stream } => {
self.await_block(block_stream).await?
}
RunnerState::ProcessingBlock {
block_stream,
block,
cursor,
} => {
has_processed_blocks = true;
self.process_block_state(block_stream, block, cursor)
.await?
}
RunnerState::Reverting {
block_stream,
to_ptr,
cursor,
} => {
self.handle_revert_state(block_stream, to_ptr, cursor)
.await?
}
RunnerState::Stopped { reason } => {
return self.finalize(reason).await;
}
};
}
}
/// Process a block and determine the next state.
///
/// This is the state machine wrapper around `process_block` that handles
/// the block processing action and determines state transitions.
async fn process_block_state(
&mut self,
block_stream: Cancelable<Box<dyn BlockStream<C>>>,
block: BlockWithTriggers<C>,
cursor: FirehoseCursor,
) -> Result<RunnerState<C>, SubgraphRunnerError> {
let block_ptr = block.ptr();
self.metrics
.stream
.deployment_head
.set(block_ptr.number as f64);
if block.trigger_count() > 0 {
self.metrics
.subgraph
.block_trigger_count
.observe(block.trigger_count() as f64);
}
// Check if we should skip this block (optimization for blocks without triggers)
if block.trigger_count() == 0
&& self.state.skip_ptr_updates_timer.elapsed() <= SKIP_PTR_UPDATES_THRESHOLD
&& !self.inputs.store.is_deployment_synced()
&& !close_to_chain_head(&block_ptr, &self.inputs.chain.chain_head_ptr().await?, 1000)
{
// Skip this block and continue with the same stream
return Ok(RunnerState::AwaitingBlock { block_stream });
} else {
self.state.skip_ptr_updates_timer = Instant::now();
}
let block_start = Instant::now();
let action = {
let stopwatch = &self.metrics.stream.stopwatch;
let _section = stopwatch.start_section(PROCESS_BLOCK_SECTION_NAME);
self.process_block(block, cursor).await
};
let action = self.handle_action(block_start, block_ptr, action).await?;
self.update_deployment_synced_metric();
if self.is_canceled() {
return self.cancel();
}
self.metrics
.subgraph
.observe_block_processed(block_start.elapsed(), action.block_finished());
// Convert Action to RunnerState
match action {
Action::Continue => Ok(RunnerState::AwaitingBlock { block_stream }),
Action::Restart => Ok(RunnerState::Restarting {
reason: RestartReason::DynamicDataSourceCreated,
}),
Action::Stop => Ok(RunnerState::Stopped {
reason: StopReason::MaxEndBlockReached,
}),
}
}
/// Handle a revert event and determine the next state.
///
/// This is the state machine wrapper around `handle_revert` that handles
/// the revert action and determines state transitions.
async fn handle_revert_state(
&mut self,
block_stream: Cancelable<Box<dyn BlockStream<C>>>,
revert_to_ptr: BlockPtr,
cursor: FirehoseCursor,
) -> Result<RunnerState<C>, SubgraphRunnerError> {
let stopwatch = &self.metrics.stream.stopwatch;
let _section = stopwatch.start_section(HANDLE_REVERT_SECTION_NAME);
let action = self.handle_revert(revert_to_ptr, cursor).await?;
match action {
Action::Continue => Ok(RunnerState::AwaitingBlock { block_stream }),
Action::Restart => Ok(RunnerState::Restarting {
reason: RestartReason::StoreError,
}),
Action::Stop => Ok(RunnerState::Stopped {
reason: StopReason::Canceled,
}),
}
}
async fn transact_block_state(
&mut self,
logger: &Logger,
block_ptr: BlockPtr,
firehose_cursor: FirehoseCursor,
block_time: BlockTime,
block_state: BlockState,
proof_of_indexing: SharedProofOfIndexing,
offchain_mods: Vec<EntityModification>,
processed_offchain_data_sources: Vec<StoredDynamicDataSource>,
) -> Result<(), ProcessingError> {
fn log_evict_stats(logger: &Logger, evict_stats: &EvictStats) {
trace!(logger, "Entity cache statistics";
"weight" => evict_stats.new_weight,
"evicted_weight" => evict_stats.evicted_weight,
"count" => evict_stats.new_count,
"evicted_count" => evict_stats.evicted_count,
"stale_update" => evict_stats.stale_update,
"hit_rate" => format!("{:.0}%", evict_stats.hit_rate_pct()),
"accesses" => evict_stats.accesses,
"evict_time_ms" => evict_stats.evict_time.as_millis());
}
let BlockState {
deterministic_errors,
persisted_data_sources,
metrics: block_state_metrics,
mut entity_cache,
..
} = block_state;
let first_error = deterministic_errors.first().cloned();
let has_errors = first_error.is_some();
// Avoid writing to store if block stream has been canceled
if self.is_canceled() {
return Err(ProcessingError::Canceled);
}
if let Some(proof_of_indexing) = proof_of_indexing.into_inner() {
update_proof_of_indexing(
proof_of_indexing,
block_time,
&self.metrics.host.stopwatch,
&mut entity_cache,
)
.await
.non_deterministic()?;
}
let section = self
.metrics
.host
.stopwatch
.start_section("as_modifications");
let ModificationsAndCache {
modifications: mut mods,
entity_lfu_cache: cache,
evict_stats,
} = entity_cache
.as_modifications(block_ptr.number, &self.metrics.host.stopwatch)
.await
.classify()?;
section.end();
log_evict_stats(&self.logger, &evict_stats);
mods.extend(offchain_mods);
// Put the cache back in the state, asserting that the placeholder cache was not used.
assert!(self.state.entity_lfu_cache.is_empty());
self.state.entity_lfu_cache = cache;
if !mods.is_empty() {
info!(&logger, "Applying {} entity operation(s)", mods.len());
}
let err_count = deterministic_errors.len();
for (i, e) in deterministic_errors.iter().enumerate() {
let message = format!("{:#}", e).replace('\n', "\t");
error!(&logger, "Subgraph error {}/{}", i + 1, err_count;
"error" => message,
"code" => LogCode::SubgraphSyncingFailure
);
}
// Transact entity operations into the store and update the
// subgraph's block stream pointer
let _section = self.metrics.host.stopwatch.start_section("transact_block");
let start = Instant::now();
// If a deterministic error has happened, make the PoI to be the only entity that'll be stored.
if has_errors && self.inputs.errors_are_fatal() {
let is_poi_entity =
|entity_mod: &EntityModification| entity_mod.key().entity_type.is_poi();
mods.retain(is_poi_entity);
// Confidence check
assert!(
mods.len() == 1,
"There should be only one PoI EntityModification"
);
}
let is_caught_up = self.is_caught_up(&block_ptr).await.non_deterministic()?;
if !self.state.postponed_indexes_created
&& close_to_chain_head(
&block_ptr,
&self.state.cached_head_ptr,
ENV_VARS.postpone_indexes_creation_threshold,
)
{
self.state.postponed_indexes_created = true;
self.inputs
.store
.create_postponed_indexes()
.await
.non_deterministic()?;
}
self.inputs
.store
.transact_block_operations(
block_ptr.clone(),
block_time,
firehose_cursor,
mods,
&self.metrics.host.stopwatch,
persisted_data_sources,
deterministic_errors,
processed_offchain_data_sources,
self.inputs.errors_are_non_fatal(),
is_caught_up,
)
.await
.classify()
.detail("Failed to transact block operations")?;
// For subgraphs with `nonFatalErrors` feature disabled, we consider
// any error as fatal.
//
// So we do an early return to make the subgraph stop processing blocks.
//
// In this scenario the only entity that is stored/transacted is the PoI,
// all of the others are discarded.
if has_errors && self.inputs.errors_are_fatal() {
if let Err(e) = self.inputs.store.flush().await {
error!(logger, "Failed to flush store after fatal errors"; "error" => format!("{:#}", e));
}
// Only the first error is reported.
return Err(ProcessingError::Deterministic(Box::new(
first_error.unwrap(),
)));
}
let elapsed = start.elapsed().as_secs_f64();
self.metrics
.subgraph
.block_ops_transaction_duration
.observe(elapsed);
block_state_metrics
.flush_metrics_to_store(logger, block_ptr, self.inputs.deployment.id)
.non_deterministic()?;
if has_errors {
self.maybe_cancel().await?;
}
Ok(())
}
/// Cancel the subgraph if `disable_fail_fast` is not set and it is not
/// synced
async fn maybe_cancel(&self) -> Result<(), ProcessingError> {
// To prevent a buggy pending version from replacing a current version, if errors are
// present the subgraph will be unassigned.
let store = &self.inputs.store;
if !ENV_VARS.disable_fail_fast && !store.is_deployment_synced() {
store
.pause_subgraph()
.await
.map_err(|e| ProcessingError::Unknown(e.into()))?;
// Use `Canceled` to avoiding setting the subgraph health to failed, an error was
// just transacted so it will be already be set to unhealthy.
Err(ProcessingError::Canceled)
} else {
Ok(())
}
}
async fn match_and_decode_many<'a, F>(
&'a self,
logger: &Logger,
block: &Arc<C::Block>,
triggers: Vec<Trigger<C>>,
hosts_filter: F,
) -> Result<Vec<RunnableTriggers<'a, C>>, MappingError>
where
F: Fn(&TriggerData<C>) -> Box<dyn Iterator<Item = &'a T::Host> + Send + 'a>,
{
let triggers = triggers.into_iter().map(|t| match t {
Trigger::Chain(t) => TriggerData::Onchain(t),
Trigger::Subgraph(t) => TriggerData::Subgraph(t),
});
self.ctx
.decoder
.match_and_decode_many(
logger,
block,
triggers,
hosts_filter,
&self.metrics.subgraph,
)
.await
}
// =========================================================================
// Pipeline Stage Methods
// =========================================================================
//
// The following methods implement the block processing pipeline stages.
// Each stage handles a specific phase of block processing:
//
// 1. match_triggers: Match and decode triggers against hosts
// 2. execute_triggers: Execute the matched triggers
// 3. process_dynamic_data_sources: Handle dynamically created data sources
// 4. (process_offchain_triggers): Existing handle_offchain_triggers method
// 5. (persist_block_state): Existing transact_block_state method
//
// =========================================================================
/// Pipeline Stage 1: Match triggers to hosts and decode them.
///
/// Takes raw triggers from a block and matches them against all registered
/// hosts, returning runnable triggers ready for execution.
async fn match_triggers<'a>(
&'a self,
logger: &Logger,
block: &Arc<C::Block>,
triggers: Vec<Trigger<C>>,
) -> Result<Vec<RunnableTriggers<'a, C>>, MappingError> {
let hosts_filter = |trigger: &TriggerData<C>| self.ctx.instance.hosts_for_trigger(trigger);
self.match_and_decode_many(logger, block, triggers, hosts_filter)
.await
}
/// Pipeline Stage 2: Execute matched triggers.
///
/// Takes runnable triggers and executes them using the TriggerRunner,
/// accumulating state changes in the block state.
async fn execute_triggers(
&self,
block: &Arc<C::Block>,
runnables: Vec<RunnableTriggers<'_, C>>,
block_state: BlockState,
proof_of_indexing: &SharedProofOfIndexing,
causality_region: &str,
) -> Result<BlockState, MappingError> {
let trigger_runner = TriggerRunner::new(
self.ctx.trigger_processor.as_ref(),
&self.logger,
&self.metrics.subgraph,
&self.inputs.debug_fork,
self.inputs.instrument,
);
trigger_runner
.execute(
block,
runnables,
block_state,
proof_of_indexing,
causality_region,
)
.await
}
/// Pipeline Stage 3: Process dynamically created data sources.
///
/// This loop processes data sources created during trigger execution:
/// 1. Instantiate the created data sources
/// 2. Reprocess triggers from this block that match the new data sources
/// 3. Repeat until no more data sources are created
///
/// Note: This algorithm processes data sources spawned on the same block
/// _breadth first_ on the tree implied by the parent-child relationship
/// between data sources.
async fn process_dynamic_data_sources(
&mut self,
logger: &Logger,
block: &Arc<C::Block>,
firehose_cursor: &FirehoseCursor,
mut block_state: BlockState,
proof_of_indexing: &SharedProofOfIndexing,
causality_region: &str,
) -> Result<BlockState, ProcessingError> {
fn log_triggers_found<C: Blockchain>(logger: &Logger, triggers: &[Trigger<C>]) {
if triggers.len() == 1 {
info!(logger, "1 trigger found in this block");
} else if triggers.len() > 1 {
info!(logger, "{} triggers found in this block", triggers.len());
}
}
let _section = self
.metrics
.stream
.stopwatch
.start_section(HANDLE_CREATED_DS_SECTION_NAME);
while block_state.has_created_data_sources() {
// Instantiate dynamic data sources, removing them from the block state.
let (data_sources, runtime_hosts) =
self.create_dynamic_data_sources(block_state.drain_created_data_sources())?;
let filter = &Arc::new(TriggerFilterWrapper::new(
C::TriggerFilter::from_data_sources(
data_sources.iter().filter_map(DataSource::as_onchain),
),
vec![],
));
// TODO: We have to pass a reference to `block` to
// `refetch_block`, otherwise the call to
// handle_offchain_triggers below gets an error that `block`
// has moved. That is extremely fishy since it means that
// `handle_offchain_triggers` uses the non-refetched block
//
// It's also not clear why refetching needs to happen inside
// the loop; will firehose really return something diffrent
// each time even though the cursor doesn't change?
let block = self.refetch_block(logger, block, firehose_cursor).await?;
// Reprocess the triggers from this block that match the new data sources
let block_with_triggers = self
.inputs
.triggers_adapter
.triggers_in_block(logger, block.as_ref().clone(), filter)
.await
.non_deterministic()?;
let triggers = block_with_triggers.trigger_data;
log_triggers_found::<C>(logger, &triggers);
// Add entity operations for the new data sources to the block state
// and add runtimes for the data sources to the subgraph instance.
self.persist_dynamic_data_sources(&mut block_state, data_sources);
// Process the triggers in each host in the same order the
// corresponding data sources have been created.
let hosts_filter = |_: &'_ TriggerData<C>| -> Box<dyn Iterator<Item = _> + Send> {
Box::new(runtime_hosts.iter().map(Arc::as_ref))
};
let runnables = self
.match_and_decode_many(logger, &block, triggers, hosts_filter)
.await;
let trigger_runner = TriggerRunner::new(
self.ctx.trigger_processor.as_ref(),
&self.logger,
&self.metrics.subgraph,
&self.inputs.debug_fork,
self.inputs.instrument,
);
let res = match runnables {
Ok(runnables) => {
trigger_runner
.execute(
&block,