fix cldeadlock crashes

- dangling clnt->thd: worker now clears sqlthd->clnt / clnt->thd (under
  gbl_sql_lock / clnt->sql_lk) before signal_clnt_as_done; skipped on nested
  replay; watchdog long-running-clnt scan reads clnt->thd once
- SEGV in newsql_write_postponed_evbuffer: only set postponed_write if the
  postponed row was actually saved; NULL-check before sending
- abort sending startgen with rqid 0: keep rqid on the lock-desired restart
  path when keep_rqid, and don't resend commit logic on an ended session

Author: markhannum

db/osqlsqlthr.c

@@ -246,9 +246,12 @@ static int osql_sock_start_int(struct sqlclntstate *clnt, int type,
     rc = clnt_check_bdb_lock_desired(clnt);
     if (rc) {
         logmsg(LOGMSG_ERROR, "recover_deadlock returned %d\n", rc);
-        rc = osql_end(clnt);
-        if (rc) {
-            logmsg(LOGMSG_ERROR, "%s failed to end osql %d\n", __func__, rc);
+        /* keep_rqid: session is being restarted, don't zero rqid */
+        if (!keep_rqid) {
+            rc = osql_end(clnt);
+            if (rc) {
+                logmsg(LOGMSG_ERROR, "%s failed to end osql %d\n", __func__, rc);
+            }
         }
         return SQLITE_BUSY;
     }
@@ -1211,7 +1214,8 @@ int osql_sock_commit(struct sqlclntstate *clnt, int type, enum trans_clntcomm si
                         int keep_session = !is_final || (get_cnonce(clnt, &snap) == 0);
 
                         rc = osql_sock_restart(clnt, 1, keep_session, is_final);
-                        if (sock_restart_retryable_rcode(rc) && !clnt->is_coordinator) {
+                        /* don't resend on an ended session (rqid 0) */
+                        if (sock_restart_retryable_rcode(rc) && osql->rqid && !clnt->is_coordinator) {
                             if (gbl_master_swing_sock_restart_sleep) {
                                 sleep(gbl_master_swing_sock_restart_sleep);
                             }

db/reqlog.c

@@ -1992,7 +1992,9 @@ void reqlog_long_running_clnt(struct sqlclntstate *clnt)
 {
     int have_fingerprint = 0;
     char fp[FINGERPRINTSZ] = {0};
-    if (clnt->done || !clnt->thd || !clnt->sql || !clnt->thd->logger) return;
+    /* single read: worker clears clnt->thd concurrently */
+    struct sqlthdstate *thd = clnt->thd;
+    if (clnt->done || !thd || !clnt->sql || !thd->logger) return;
 
     if (can_consume(clnt) == 1) {
         return; /* Do not log consumers */

db/sqlglue.c

@@ -10268,7 +10268,11 @@ int recover_deadlock_flags(bdb_state_type *bdb_state, struct sqlclntstate *clnt,
         clnt->recover_deadlock_thd = pthread_self();
         comdb2_cheapstack_char_array(clnt->recover_deadlock_stack, RECOVER_DEADLOCK_MAX_STACK);
 #endif
-        recover_deadlock_sc_cleanup(clnt->thd->sqlthd);
+        /* use TLS thd (as recover_deadlock_flags_int does): clnt->thd may
+         * already be cleared when called from the post-done flush path */
+        struct sql_thread *sqlthd = pthread_getspecific(query_info_key);
+        if (sqlthd)
+            recover_deadlock_sc_cleanup(sqlthd);
         assert(bdb_lockref() == 0);
     } else {
         assert(bdb_lockref() > 0);

db/sqlinterfaces.c

@@ -4081,8 +4081,8 @@ static int run_stmt(struct sqlthdstate *thd, struct sqlclntstate *clnt,
             if (rc)
                 return rc;
         } else {
-            postponed_write = 1;
-            send_row(clnt, stmt, row_id, 1, NULL);
+            /* only claim a postponed row if the save succeeded */
+            postponed_write = (send_row(clnt, stmt, row_id, 1, NULL) == 0);
         }
 
         rowcount++;
@@ -4858,6 +4858,21 @@ static int can_execute_sql_query_now(
   return 1;
 }
 
+/* clear before signal: signal hands clnt back to the event thread.
+ * nested replay call: outer frame still owns thd */
+static void clnt_detach_thd(struct sqlclntstate *clnt, struct sql_thread *sqlthd)
+{
+    if (clnt->osql.in_replay_nested)
+        return;
+    Pthread_mutex_lock(&gbl_sql_lock);
+    sqlthd->clnt = NULL;
+    Pthread_mutex_unlock(&gbl_sql_lock);
+    /* sql_lk: watchdog reads clnt->thd under it */
+    Pthread_mutex_lock(&clnt->sql_lk);
+    clnt->thd = NULL; /* thd is about to go away */
+    Pthread_mutex_unlock(&clnt->sql_lk);
+}
+
 void sqlengine_work_appsock(struct sqlthdstate *thd, struct sqlclntstate *clnt)
 {
     struct sql_thread *sqlthd = thd->sqlthd;
@@ -4908,6 +4923,7 @@ void sqlengine_work_appsock(struct sqlthdstate *thd, struct sqlclntstate *clnt)
         clnt->osql.timings.query_finished = osql_log_time();
         osql_log_time_done(clnt);
         clnt_change_state(clnt, CONNECTION_IDLE);
+        clnt_detach_thd(clnt, sqlthd);
         signal_clnt_as_done(clnt);
         return;
     }
@@ -4960,6 +4976,7 @@ void sqlengine_work_appsock(struct sqlthdstate *thd, struct sqlclntstate *clnt)
     osql_log_time_done(clnt);
     clnt_change_state(clnt, CONNECTION_IDLE);
     debug_close_clnt(clnt);
+    clnt_detach_thd(clnt, sqlthd);
     signal_clnt_as_done(clnt);
 
     thrman_setid(thrman_self(), "[done]");

db/sqloffload.c

@@ -641,7 +641,9 @@ int osql_clean_sqlclntstate(struct sqlclntstate *clnt)
             abort();
     }
 
+    int in_replay_nested = osql->in_replay_nested; /* call-stack state, not txn state */
     bzero(osql, sizeof(*osql));
+    osql->in_replay_nested = in_replay_nested;
     listc_init(&osql->shadtbls, offsetof(struct shad_tbl, linkv));
 
     sql_set_sqlengine_state(clnt, __FILE__, __LINE__, SQLENG_NORMAL_PROCESS);

plugins/newsql/newsql.c

@@ -616,6 +616,8 @@ static int newsql_save_postponed_row(struct sqlclntstate *clnt,
 static int newsql_send_postponed_row(struct sqlclntstate *clnt)
 {
     struct newsql_appdata *appdata = clnt->appdata;
+    if (appdata->postponed == NULL)
+        return -1;
     return appdata->write_postponed(clnt);
 }