store message first, remove on ack

xwords4/relay/cref.cpp

@@ -875,13 +875,13 @@ putNetShort( uint8_t** bufpp, unsigned short s )
     *bufpp += sizeof(s);
 }
 
-void
+int
 CookieRef::store_message( HostID dest, const uint8_t* buf, 
                           unsigned int len )
 {
     logf( XW_LOGVERBOSE0, "%s: storing msg size %d for dest %d", __func__,
           len, dest );
-    DBMgr::Get()->StoreMessage( ConnName(), dest, buf, len );
+    return DBMgr::Get()->StoreMessage( ConnName(), dest, buf, len );
 }
 
 void
@@ -1044,6 +1044,7 @@ CookieRef::postCheckAllHere()
 void
 CookieRef::postDropDevice( HostID hostID )
 {
+    logf( XW_LOGINFO, "%s(hostID=%d)", __func__, hostID );
     CRefEvent evt( XWE_ACKTIMEOUT );
     evt.u.ack.srcID = hostID;
     m_eventQueue.push_back( evt );
@@ -1192,21 +1193,16 @@ CookieRef::sendAnyStored( const CRefEvent* evt )
 }
 
 typedef struct _StoreData {
-    string connName;
+    int msgID;
-    HostID dest;
-    uint8_t* buf;
-    int buflen;
 } StoreData;
 
 void
 CookieRef::storeNoAck( bool acked, uint32_t packetID, void* data )
 {
     StoreData* sdata = (StoreData*)data;
-    if ( !acked ) {
+    if ( acked ) {
-        DBMgr::Get()->StoreMessage( sdata->connName.c_str(), sdata->dest, 
+        DBMgr::Get()->RemoveStoredMessages( &sdata->msgID, 1 );
-                                    sdata->buf, sdata->buflen );
     }
-    free( sdata->buf );
     delete sdata;
 }
 
@@ -1237,17 +1233,13 @@ CookieRef::forward_or_store( const CRefEvent* evt )
         }
 
         uint32_t packetID = 0;
+        int msgID = store_message( dest, buf, buflen );
         if ( (NULL == destAddr)
              || !send_with_length( destAddr, dest, buf, buflen, true, 
                                    &packetID ) ) {
-            store_message( dest, buf, buflen );
+        } else if ( 0 != msgID && 0 != packetID ) { // sent via UDP
-        } else if ( 0 != packetID ) { // sent via UDP
             StoreData* data = new StoreData;
-            data->connName = m_connName;
+            data->msgID = msgID;
-            data->dest = dest;
-            data->buf = (uint8_t*)malloc( buflen );
-            memcpy( data->buf, buf, buflen );
-            data->buflen = buflen;
             UDPAckTrack::setOnAck( storeNoAck, packetID, data );
         }
 
@@ -1376,20 +1368,16 @@ CookieRef::sendAllHere( bool initial )
        through the vector each time. */
     HostID dest;
     for ( dest = 1; dest <= m_nPlayersSought; ++dest ) {
-        bool sent = false;
         *idLoc = dest;   /* write in this target's hostId */
 
         {
             RWReadLock rrl( &m_socketsRWLock );
             HostRec* hr = m_sockets[dest-1];
             if ( !!hr ) {
-                sent = send_with_length( &hr->m_addr, dest, buf, 
+                (void)send_with_length( &hr->m_addr, dest, buf, bufp-buf, true );
-                                         bufp-buf, true );
             }
         }
-        if ( !sent ) {
+        (void)store_message( dest, buf, bufp-buf );
-            store_message( dest, buf, bufp-buf );
-        }
     }
 } /* sendAllHere */
 

xwords4/relay/cref.h

@@ -275,8 +275,7 @@ class CookieRef {
 
     bool notInUse(void) { return m_cid == 0; }
 
-    void store_message( HostID dest, const uint8_t* buf, 
+    int store_message( HostID dest, const uint8_t* buf, unsigned int len );
-                        unsigned int len );
     void send_stored_messages( HostID dest, const AddrInfo* addr );
 
     void printSeeds( const char* caller );

xwords4/relay/dbmgr.cpp

@@ -1014,15 +1014,16 @@ DBMgr::CountStoredMessages( DevIDRelay relayID )
     return getCountWhere( MSGS_TABLE, test );
 }
 
-void
+int
 DBMgr::StoreMessage( DevIDRelay destDevID, const uint8_t* const buf,
                      int len )
 {
+    int msgID = 0;
     clearHasNoMessages( destDevID );
 
     size_t newLen;
     const char* fmt = "INSERT INTO " MSGS_TABLE " "
-        "(devid, %s, msglen) VALUES(%d, %s'%s', %d)";
+        "(devid, %s, msglen) VALUES(%d, %s'%s', %d) RETURNING id";
     
     StrWPF query;
     if ( m_useB64 ) {
@@ -1038,13 +1039,20 @@ DBMgr::StoreMessage( DevIDRelay destDevID, const uint8_t* const buf,
     }
 
     logf( XW_LOGINFO, "%s: query: %s", __func__, query.c_str() );
-    execSql( query );
+
+    PGresult* result = PQexec( getThreadConn(), query.c_str() );
+    if ( 1 == PQntuples( result ) ) {
+        msgID = atoi( PQgetvalue( result, 0, 0 ) );
+    }
+    PQclear( result );
+    return msgID;
 }
 
-void
+int
 DBMgr::StoreMessage( const char* const connName, int destHid,
                      const uint8_t* buf, int len )
 {
+    int msgID = 0;
     clearHasNoMessages( connName, destHid );
 
     DevIDRelay devID = getDevID( connName, destHid );
@@ -1074,7 +1082,7 @@ DBMgr::StoreMessage( const char* const connName, int destHid,
 #ifdef HAVE_STIME
                     " AND stime='epoch'" 
 #endif
-                    " );", connName, destHid, b64 );
+                    " )", connName, destHid, b64 );
         g_free( b64 );
     } else {
         uint8_t* bytes = PQescapeByteaConn( getThreadConn(), buf, 
@@ -1085,9 +1093,17 @@ DBMgr::StoreMessage( const char* const connName, int destHid,
                     "E", bytes, len );
         PQfreemem( bytes );
     }
+    query.catf(" RETURNING id;");
 
     logf( XW_LOGINFO, "%s: query: %s", __func__, query.c_str() );
-    execSql( query );
+    PGresult* result = PQexec( getThreadConn(), query.c_str() );
+    if ( 1 == PQntuples( result ) ) {
+        msgID = atoi( PQgetvalue( result, 0, 0 ) );
+    } else {
+        logf( XW_LOGINFO, "Not stored; duplicate?" );
+    }
+    PQclear( result );
+    return msgID;
 }
 
 void

xwords4/relay/dbmgr.h

@@ -137,9 +137,9 @@ class DBMgr {
     /* message storage -- different DB */
     int CountStoredMessages( const char* const connName );
     int CountStoredMessages( DevIDRelay relayID );
-    void StoreMessage( DevIDRelay destRelayID, const uint8_t* const buf,
+    int StoreMessage( DevIDRelay destRelayID, const uint8_t* const buf,
                       int len );
-    void StoreMessage( const char* const connName, int destHid,
+    int StoreMessage( const char* const connName, int destHid,
                       const uint8_t* const buf, int len );
     void GetStoredMessages( DevIDRelay relayID, vector<MsgInfo>& msgs );
     void GetStoredMessages( const char* const connName, HostID hid, 

xwords4/relay/xwrelay.cpp

@@ -761,13 +761,17 @@ send_havemsgs( const AddrInfo* addr )
 class MsgClosure {
 public:
     MsgClosure( DevIDRelay dest, const vector<uint8_t>* packet,
-                OnMsgAckProc proc, void* procClosure )
+                int msgID, OnMsgAckProc proc, void* procClosure )
     {
+        assert(m_msgID != 0);
         m_destDevID = dest;
         m_packet = *packet;
         m_proc = proc;
         m_procClosure = procClosure;
+        m_msgID = msgID;
     }
+    int getMsgID() { return m_msgID; }
+    int m_msgID;
     DevIDRelay m_destDevID;
     vector<uint8_t> m_packet;
     OnMsgAckProc m_proc;
@@ -778,9 +782,14 @@ static void
 onPostedMsgAcked( bool acked, uint32_t packetID, void* data )
 {
     MsgClosure* mc = (MsgClosure*)data;
-    if ( !acked ) {
+    int msgID = mc->getMsgID();
-        DBMgr::Get()->StoreMessage( mc->m_destDevID, mc->m_packet.data(),
+    if ( acked ) {
-                                    mc->m_packet.size() );
+        DBMgr::Get()->RemoveStoredMessages( &msgID, 1 );
+    } else {
+        assert( msgID != 0 );
+        // So we only store after ack fails? Change that!!!
+        // DBMgr::Get()->StoreMessage( mc->m_destDevID, mc->m_packet.data(),
+        //                             mc->m_packet.size() );
     }
     if ( NULL != mc->m_proc ) {
         (*mc->m_proc)( acked, mc->m_destDevID, packetID, mc->m_procClosure );
@@ -793,6 +802,8 @@ static bool
 post_or_store( DevIDRelay destDevID, vector<uint8_t>& packet, uint32_t packetID,
                OnMsgAckProc proc, void* procClosure )
 {
+    int msgID = DBMgr::Get()->StoreMessage( destDevID, packet.data(), packet.size() );
+
     const AddrInfo::AddrUnion* addru = DevMgr::Get()->get( destDevID );
     bool canSendNow = !!addru;
     
@@ -804,16 +815,13 @@ post_or_store( DevIDRelay destDevID, vector<uint8_t>& packet, uint32_t packetID,
         if ( get_addr_info_if( &addr, &sock, &dest_addr ) ) {
             sent = 0 < send_packet_via_udp_impl( packet, sock, dest_addr );
 
-            if ( sent ) {
+            if ( sent && msgID != 0 ) {
-                MsgClosure* mc = new MsgClosure( destDevID, &packet,
+                MsgClosure* mc = new MsgClosure( destDevID, &packet, msgID,
                                                  proc, procClosure );
                 UDPAckTrack::setOnAck( onPostedMsgAcked, packetID, (void*)mc );
             }
         }
     }
-    if ( !sent ) {
-        DBMgr::Get()->StoreMessage( destDevID, packet.data(), packet.size() );
-    }
     return sent;
 }