MSC4242: State DAGs (CSAPI)

changelog.d/19424.feature

@@ -0,0 +1 @@
+Add experimental support for [MSC4242](https://github.com/matrix-org/matrix-spec-proposals/pull/4242): State DAGs. Excludes federation support.

contrib/grafana/synapse.json

@@ -6809,6 +6809,155 @@
           ],
           "title": "Stale extremity dropping",
           "type": "timeseries"
+        },
+        {
+          "datasource": {
+            "uid": "${DS_PROMETHEUS}"
+          },
+          "description": "For a given percentage P, the number X where P% of events were persisted to rooms with X state DAG forward extremities or fewer.",
+          "fieldConfig": {
+            "defaults": {
+              "links": []
+            },
+            "overrides": []
+          },
+          "gridPos": {
+            "h": 8,
+            "w": 12,
+            "x": 12,
+            "y": 50
+          },
+          "id": 181,
+          "options": {
+            "alertThreshold": true
+          },
+          "pluginVersion": "9.2.2",
+          "targets": [
+            {
+              "datasource": {
+                "uid": "${DS_PROMETHEUS}"
+              },
+              "expr": "histogram_quantile(0.5, rate(synapse_storage_msc4242_state_dag_forward_extremities_persisted_bucket{server_name=\"$server_name\"}[$bucket_size]) and on (instance, job, index) (synapse_storage_events_persisted_events_total > 0))",
+              "format": "time_series",
+              "intervalFactor": 1,
+              "legendFormat": "50%",
+              "refId": "A"
+            },
+            {
+              "datasource": {
+                "uid": "${DS_PROMETHEUS}"
+              },
+              "expr": "histogram_quantile(0.75, rate(synapse_storage_msc4242_state_dag_forward_extremities_persisted_bucket{server_name=\"$server_name\"}[$bucket_size]) and on (instance, job, index) (synapse_storage_events_persisted_events_total > 0))",
+              "format": "time_series",
+              "intervalFactor": 1,
+              "legendFormat": "75%",
+              "refId": "B"
+            },
+            {
+              "datasource": {
+                "uid": "${DS_PROMETHEUS}"
+              },
+              "expr": "histogram_quantile(0.90, rate(synapse_storage_msc4242_state_dag_forward_extremities_persisted_bucket{server_name=\"$server_name\"}[$bucket_size]) and on (instance, job, index) (synapse_storage_events_persisted_events_total > 0))",
+              "format": "time_series",
+              "intervalFactor": 1,
+              "legendFormat": "90%",
+              "refId": "C"
+            },
+            {
+              "datasource": {
+                "uid": "${DS_PROMETHEUS}"
+              },
+              "expr": "histogram_quantile(0.99, rate(synapse_storage_msc4242_state_dag_forward_extremities_persisted_bucket{server_name=\"$server_name\"}[$bucket_size]) and on (instance, job, index) (synapse_storage_events_persisted_events_total > 0))",
+              "format": "time_series",
+              "intervalFactor": 1,
+              "legendFormat": "99%",
+              "refId": "D"
+            }
+          ],
+          "title": "Events persisted, by number of state DAG forward extremities in room (quantiles)",
+          "type": "timeseries"
+        },
+        {
+          "datasource": {
+            "uid": "${DS_PROMETHEUS}"
+          },
+          "description": "Colour reflects the number of events persisted to rooms with the given number of state DAG forward extremities, or fewer.",
+          "fieldConfig": {
+            "defaults": {
+              "custom": {
+                "hideFrom": {
+                  "legend": false,
+                  "tooltip": false,
+                  "viz": false
+                },
+                "scaleDistribution": {
+                  "type": "linear"
+                }
+              }
+            },
+            "overrides": []
+          },
+          "gridPos": {
+            "h": 8,
+            "w": 12,
+            "x": 0,
+            "y": 50
+          },
+          "id": 127,
+          "options": {
+            "calculate": false,
+            "calculation": {},
+            "cellGap": 1,
+            "cellValues": {},
+            "color": {
+              "exponent": 0.5,
+              "fill": "#5794F2",
+              "min": 0,
+              "mode": "opacity",
+              "reverse": false,
+              "scale": "exponential",
+              "scheme": "Oranges",
+              "steps": 128
+            },
+            "exemplars": {
+              "color": "rgba(255,0,255,0.7)"
+            },
+            "filterValues": {
+              "le": 1e-9
+            },
+            "legend": {
+              "show": true
+            },
+            "rowsFrame": {
+              "layout": "auto"
+            },
+            "showValue": "never",
+            "tooltip": {
+              "show": true,
+              "yHistogram": true
+            },
+            "yAxis": {
+              "axisPlacement": "left",
+              "decimals": 0,
+              "reverse": false,
+              "unit": "short"
+            }
+          },
+          "pluginVersion": "9.2.2",
+          "targets": [
+            {
+              "datasource": {
+                "uid": "${DS_PROMETHEUS}"
+              },
+              "expr": "rate(synapse_storage_msc4242_state_dag_forward_extremities_persisted_bucket{server_name=\"$server_name\"}[$bucket_size]) and on (instance, job, index) (synapse_storage_events_persisted_events_total > 0)",
+              "format": "heatmap",
+              "intervalFactor": 1,
+              "legendFormat": "{{le}}",
+              "refId": "A"
+            }
+          ],
+          "title": "Events persisted, by number of state DAG forward extremities in room (heatmap)",
+          "type": "heatmap"
         }
       ],
       "title": "Extremities",
@@ -7711,4 +7860,4 @@
   "uid": "000000012",
   "version": 1,
   "weekStart": ""
-}
+}

rust/src/events/internal_metadata.rs

@@ -65,6 +65,7 @@ enum EventInternalMetadataData {
     DelayId(Box<str>),
     TokenId(i64),
     DeviceId(Box<str>),
+    CalculatedAuthEventIDs(Vec<String>), // MSC4242: State DAGs
 }
 
 impl EventInternalMetadataData {
@@ -140,6 +141,10 @@ impl EventInternalMetadataData {
                 pyo3::intern!(py, "device_id"),
                 o.into_pyobject(py).unwrap_infallible().into_any(),
             ),
+            EventInternalMetadataData::CalculatedAuthEventIDs(o) => (
+                pyo3::intern!(py, "calculated_auth_event_ids"),
+                o.into_pyobject(py).unwrap().into_any(),
+            ),
         }
     }
 
@@ -218,6 +223,11 @@ impl EventInternalMetadataData {
                     .map(String::into_boxed_str)
                     .with_context(|| format!("'{key_str}' has invalid type"))?,
             ),
+            "calculated_auth_event_ids" => EventInternalMetadataData::CalculatedAuthEventIDs(
+                value
+                    .extract()
+                    .with_context(|| format!("'{key_str}' has invalid type"))?,
+            ),
             _ => return Ok(None),
         };
 
@@ -395,6 +405,10 @@ impl EventInternalMetadataInner {
         get_property_opt!(self, DelayId).map(|s| s.deref())
     }
 
+    pub fn get_calculated_auth_event_ids(&self) -> Option<&Vec<String>> {
+        get_property_opt!(self, CalculatedAuthEventIDs)
+    }
+
     pub fn get_token_id(&self) -> Option<i64> {
         get_property_opt!(self, TokenId).copied()
     }
@@ -456,6 +470,10 @@ impl EventInternalMetadataInner {
     pub fn set_device_id(&mut self, obj: String) {
         set_property!(self, DeviceId, obj.into_boxed_str());
     }
+
+    pub fn set_calculated_auth_event_ids(&mut self, obj: Vec<String>) {
+        set_property!(self, CalculatedAuthEventIDs, obj);
+    }
 }
 
 #[pyclass(frozen)]
@@ -722,6 +740,21 @@ impl EventInternalMetadata {
         Ok(())
     }
 
+    /// The calculated auth event IDs, if it was set when the event was created.
+    #[getter]
+    fn get_calculated_auth_event_ids(&self) -> PyResult<Vec<String>> {
+        let guard = self.read_inner()?;
+        attr_err(
+            guard.get_calculated_auth_event_ids().cloned(),
+            "calculated_auth_event_ids",
+        )
+    }
+    #[setter]
+    fn set_calculated_auth_event_ids(&self, obj: Vec<String>) -> PyResult<()> {
+        self.write_inner()?.set_calculated_auth_event_ids(obj);
+        Ok(())
+    }
+
     /// The delay ID, set only if the event was a delayed event.
     #[getter]
     fn get_delay_id(&self) -> PyResult<String> {