Request additional sync point on discard.

docker/prep_stmts/.gitignore

@@ -0,0 +1,2 @@
+pstst
+psmst

sources/CMakeLists.txt

@@ -24,6 +24,7 @@ set(od_src
     deploy.c
     reset.c
     frontend.c
+    backend_sync.c
     backend.c
     instance.c
     main.c

sources/backend_sync.c

@@ -0,0 +1,34 @@
+/*
+ * Odyssey.
+ *
+ * Scalable PostgreSQL connection pooler.
+ */
+
+#include <kiwi.h>
+#include <machinarium.h>
+#include <odyssey.h>
+
+int od_backend_request_sync_point(od_server_t *server)
+{
+	od_instance_t *instance = server->global->instance;
+	int rc;
+
+	machine_msg_t *msg;
+	msg = kiwi_fe_write_sync(NULL);
+	if (msg == NULL) {
+		return -1;
+	}
+	rc = od_write(&server->io, msg);
+	if (rc == -1) {
+		od_error(&instance->logger, "sync-point", server->client,
+			 server, "write error: %s", od_io_error(&server->io));
+		return NOT_OK_RESPONSE;
+	}
+
+	/* update server sync state */
+	od_server_sync_request(server, 1);
+
+	return od_backend_ready_wait(server, "sync-point", 1 /*count*/,
+				     UINT32_MAX /* timeout */,
+				     0 /*ignore error?*/);
+}

sources/backend_sync.h

@@ -0,0 +1,12 @@
+#ifndef ODYSSEY_BACKEND_SYNC_H
+#define ODYSSEY_BACKEND_SYNC_H
+
+/*
+ * Odyssey.
+ *
+ * Scalable PostgreSQL connection pooler.
+ */
+
+int od_backend_request_sync_point(od_server_t *);
+
+#endif /* ODYSSEY_BACKEND_SYNC_H */

sources/reset.c

@@ -112,6 +112,21 @@ int od_reset(od_server_t *server)
 		assert(od_server_synchronized(server));
 		break;
 	}
+
+	/* Request one more sync point here.
+	* In `od_server_synchronized` we
+	* count number of sync/query msg send to connection
+	* and number of RFQ received, if this numbers are equal,  
+	* we decide server connection as sync. However, this might be 
+	* not true, if client-server relay advanced some extended proto
+	* msgs without sync. To safely execute discard queries, we need to
+	* advadance sync point first.
+	*/
+
+	if (od_backend_request_sync_point(server) == NOT_OK_RESPONSE) {
+		goto error;
+	}
+
 	od_debug(&instance->logger, "reset", server->client, server,
 		 "synchronized");