store: Work around mysterious write errors

Author: lutter

graph/src/components/store/write.rs

@@ -933,6 +933,18 @@ impl<'a> WriteChunk<'a> {
             count: 0,
         }
     }
+
+    /// Return a vector of `WriteChunk`s each containing a single write
+    pub fn as_single_writes(&self) -> Vec<Self> {
+        (0..self.len())
+            .into_iter()
+            .map(|position| WriteChunk {
+                group: self.group,
+                chunk_size: 1,
+                position: self.position + position,
+            })
+            .collect()
+    }
 }
 
 impl<'a> IntoIterator for &WriteChunk<'a> {

store/postgres/src/deployment_store.rs

@@ -379,7 +379,7 @@ impl DeploymentStore {
             section.end();
 
             let section = stopwatch.start_section("apply_entity_modifications_insert");
-            layout.insert(conn, group, stopwatch).await?;
+            layout.insert(&self.logger, conn, group, stopwatch).await?;
             section.end();
         }
 

store/postgres/src/relational.rs

@@ -736,6 +736,7 @@ impl Layout {
 
     pub async fn insert<'a>(
         &'a self,
+        logger: &Logger,
         conn: &mut AsyncPgConnection,
         group: &'a RowGroup,
         stopwatch: &StopwatchMetrics,
@@ -769,13 +770,39 @@ impl Layout {
         for chunk in group.write_chunks(chunk_size) {
             // Empty chunks would lead to invalid SQL
             if !chunk.is_empty() {
-                InsertQuery::new(table, &chunk)?
-                    .execute(conn)
-                    .await
-                    .map_err(|e| {
+                if let Err(e) = InsertQuery::new(table, &chunk)?.execute(conn).await {
+                    // We occasionally get these errors but it's entirely
+                    // unclear what causes them. We work around that by
+                    // switching to row-by-row inserts until we can figure
+                    // out what the underlying cause is
+                    let err_msg = e.to_string();
+                    if !err_msg.contains("value too large to transmit") {
                         let (block, msg) = chunk_details(&chunk);
-                        StoreError::write_failure(e, table.object.as_str(), block, msg)
-                    })?;
+                        return Err(StoreError::write_failure(
+                            e,
+                            table.object.as_str(),
+                            block,
+                            msg,
+                        ));
+                    }
+                    let (block, msg) = chunk_details(&chunk);
+                    warn!(logger, "Insert of entire chunk failed. Trying row by row insert.";
+                        "table" => table.object.as_str(),
+                        "block" => block,
+                        "error" => err_msg,
+                        "details" => msg
+                    );
+                    for single_chunk in chunk.as_single_writes() {
+                        InsertQuery::new(table, &single_chunk)?
+                            .execute(conn)
+                            .await
+                            .map_err(|e| {
+                                let (block, msg) = chunk_details(&single_chunk);
+                                let msg = format!("{}: offending row {:?}", msg, single_chunk);
+                                StoreError::write_failure(e, table.object.as_str(), block, msg)
+                            })?;
+                    }
+                }
             }
         }
         Ok(())

store/test-store/tests/postgres/relational.rs

@@ -271,7 +271,7 @@ async fn insert_entity_at(
     );
     let group = row_group_insert(&entity_type, block, entities_with_keys_owned.clone());
     layout
-        .insert(conn, &group, &MOCK_STOPWATCH)
+        .insert(&LOGGER, conn, &group, &MOCK_STOPWATCH)
         .await
         .expect(&errmsg);
     assert_eq!(

store/test-store/tests/postgres/relational_bytes.rs

@@ -134,7 +134,7 @@ async fn insert_entity(
     let group = row_group_insert(&entity_type, 0, entities);
     let errmsg = format!("Failed to insert entity {}[{}]", entity_type, key.entity_id);
     layout
-        .insert(conn, &group, &MOCK_STOPWATCH)
+        .insert(&LOGGER, conn, &group, &MOCK_STOPWATCH)
         .await
         .expect(&errmsg);
 }