add repair-aware receiver sync policy

doc/NormDeveloperGuide.xml

@@ -885,9 +885,11 @@
 
         <para><programlisting>enum <link linkend="NormSyncPolicy"><literal>NormSyncPolicy</literal></link>
 {
-    <literal>NORM_SYNC_CURRENT</literal><literal>,
-    NORM_</literal>SYNC_ALL<literal>
-};</literal></programlisting></para>
+    <literal>NORM_SYNC_CURRENT,
+    NORM_SYNC_STREAM,
+    NORM_SYNC_ALL,
+    NORM_SYNC_REPAIR</literal>
+};</programlisting></para>
 
         <para>The use and interpretation of these values is given in the
         descriptions of the <link
@@ -5286,7 +5288,7 @@ void <link linkend="NormSetDefaultSyncPolicy"><literal>NormSetDefaultSyncPolicy<
           is the behavior observed by the receiver with regards to what
           objects it attempts to reliably receive (via transmissions of
           Negative Acknowledgements to the sender(s) or group as needed).
-          There are currently two synchronization policy types defined:</para>
+          There are currently four synchronization policy types defined:</para>
 
           <informaltable frame="all">
             <tgroup cols="2">
@@ -5298,13 +5300,29 @@ void <link linkend="NormSetDefaultSyncPolicy"><literal>NormSetDefaultSyncPolicy<
                   only. (default)</para></entry>
                 </row>
 
+                <row>
+                  <entry><para><literal>NORM_SYNC_STREAM</literal></para></entry>
+
+                  <entry><para>Synchronize to the current stream while
+                  attempting recovery from the beginning of its retained
+                  receive window.</para></entry>
+                </row>
+
                 <row>
                   <entry><para><literal>NORM_SYNC_ALL</literal></para></entry>
 
                   <entry><para>Attempt recovery and reliable reception of all
                   objects held in sender transmit object cache and newer
                   objects.</para></entry>
                 </row>
+
+                <row>
+                  <entry><para><literal>NORM_SYNC_REPAIR</literal></para></entry>
+
+                  <entry><para>Attempt reception of the observed current object
+                  and newer objects, allowing repair INFO or first-block DATA
+                  to establish the initial synchronization point.</para></entry>
+                </row>
               </tbody>
             </tgroup>
           </informaltable>
@@ -5327,6 +5345,16 @@ void <link linkend="NormSetDefaultSyncPolicy"><literal>NormSetDefaultSyncPolicy<
           the <constant>NORM_SYNC_CURRENT</constant> policy is typically
           recommended.</para>
 
+          <para>The <constant>NORM_SYNC_REPAIR</constant> policy uses the same
+          current-object receive window as
+          <constant>NORM_SYNC_CURRENT</constant>, but permits repair
+          <literal>NORM_INFO</literal> or repair <literal>NORM_DATA</literal>
+          from the first FEC block to establish synchronization. It is useful
+          for retained-object and rateless transfers where a receiver should
+          recover the object already in progress without requesting older
+          objects from the sender cache. FTI may arrive in INFO or DATA; this
+          policy does not require in-band FTI.</para>
+
           <para>The <constant>NORM_SYNC_ALL</constant> policy allows newly
           joining receivers much more aggressive behavior as they will
           immediately NACK for all objects from the "current" object backwards

include/normApi.h

@@ -146,7 +146,8 @@ typedef enum NormSyncPolicy
 {
     NORM_SYNC_CURRENT,  // attempt to receiver current/new objects only, join mid-stream
     NORM_SYNC_STREAM,   // sync to current stream, but to beginning of stream
-    NORM_SYNC_ALL       // attempt to receive old and new objects
+    NORM_SYNC_ALL,      // attempt to receive old and new objects
+    NORM_SYNC_REPAIR    // accept current-object repair traffic as an initial sync point
 } NORM_API_LINKAGE NormSyncPolicy;
 
 typedef enum NormRepairBoundary

include/normNode.h

@@ -317,7 +317,8 @@ class NormSenderNode : public NormNode, public ProtoTree::Item
         {
             SYNC_CURRENT,  // sync to detect transmit point, iff NORM_DATA from first FEC block unless stream
             SYNC_STREAM,   // same as SYNC_CURRENT, but attempts to recover stream block zero
-            SYNC_ALL      // permiscuously sync as far back as possible given rx cache size
+            SYNC_ALL,      // promiscuously sync as far back as possible given rx cache size
+            SYNC_REPAIR    // sync current object from original or repair traffic, without historical rewind
         };
 
         NormSenderNode(class NormSession& theSession, NormNodeId nodeId);  

src/common/normApp.cpp

@@ -207,6 +207,7 @@ class NormApp : public NormController, public ProtoApp
         bool                process_aborted_files;
         bool                preallocate_sender;
         NormSenderNode::RepairBoundary repair_boundary;
+        NormSenderNode::SyncPolicy sync_policy;
 
         // Debug parameters
         bool                tracing;
@@ -248,7 +249,8 @@ NormApp::NormApp()
    acking_flushes(false), watermark_pending(false), rx_buffer_size(1024*1024), rx_sock_buffer_size(0),
    rx_cache_path(NULL), post_processor(NULL), unicast_nacks(false), silent_receiver(false), 
    low_delay(false), realtime(false), rx_robust_factor(NormSession::DEFAULT_ROBUST_FACTOR), rx_persistent(true), process_aborted_files(false),
-   preallocate_sender(false), repair_boundary(NormSenderNode::BLOCK_BOUNDARY), tracing(false), tx_loss(0.0), rx_loss(0.0)
+   preallocate_sender(false), repair_boundary(NormSenderNode::BLOCK_BOUNDARY),
+   sync_policy(NormSenderNode::SYNC_CURRENT), tracing(false), tx_loss(0.0), rx_loss(0.0)
 {
     control_pipe.SetListener(this, &NormApp::OnControlEvent);
     control_pipe.SetNotifier(&GetSocketNotifier());
@@ -343,6 +345,7 @@ const char* const NormApp::cmd_list[] =
     "+moutput",      // recv message stream output
     "+sendfile",     // file/directory list to transmit
     "+info",         // 'on' | 'off' to enable/disable file info (name) transmission (default = 'on')
+    "+sync",         // receiver synchronization policy: 'current', 'stream', 'repair', or 'all'
     "+interval",     // delay time (sec) between files (0.0 sec default)
     "+repeatcount",  // How many times to repeat the file/directory list tx
     "+rinterval",    // Interval (sec) between file/directory list repeats
@@ -416,6 +419,7 @@ void NormApp::ShowHelp()
         "   +moutput,      // receiver message stream output\n"
         "   +sendfile,     // file/directory list to transmit\n"
         "   +info,         // 'on' | 'off' to enable/disable file info (name) transmission (default = 'on')\n"
+        "   +sync,         // receiver synchronization policy: 'current', 'stream', 'repair', or 'all'\n"
         "   +interval,     // delay time (sec) between files (0.0 sec default)\n"
         "   +repeatcount,  // How many times to repeat the file/directory list tx\n"
         "   +rinterval,    // Interval (sec) between file/directory list repeats\n"
@@ -883,6 +887,23 @@ bool NormApp::OnCommand(const char* cmd, const char* val)
             return false;
         }
     }
+    else if (!strncmp("sync", cmd, len))
+    {
+        if (!strcmp("current", val))
+            sync_policy = NormSenderNode::SYNC_CURRENT;
+        else if (!strcmp("stream", val))
+            sync_policy = NormSenderNode::SYNC_STREAM;
+        else if (!strcmp("repair", val))
+            sync_policy = NormSenderNode::SYNC_REPAIR;
+        else if (!strcmp("all", val))
+            sync_policy = NormSenderNode::SYNC_ALL;
+        else
+        {
+            PLOG(PL_FATAL, "NormApp::OnCommand(sync) invalid policy!\n");
+            return false;
+        }
+        if (NULL != session) session->ReceiverSetDefaultSyncPolicy(sync_policy);
+    }
     else if (!strncmp("interval", cmd, len))
     {
         if (1 != sscanf(val, "%lf", &tx_object_interval))
@@ -2383,6 +2404,7 @@ bool NormApp::OnStartup(int argc, const char*const* argv)
                 session->RcvrSetRealtime(true);           
             session->SetRxRobustFactor(rx_robust_factor);
             session->ReceiverSetDefaultRepairBoundary(repair_boundary);
+            session->ReceiverSetDefaultSyncPolicy(sync_policy);
             if (!session->StartReceiver(rx_buffer_size))
             {
                 PLOG(PL_FATAL, "NormApp::OnStartup() start receiver error!\n");

src/common/normNode.cpp

@@ -1782,7 +1782,8 @@ void NormSenderNode::HandleObjectMessage(const NormObjectMsg& msg)
                                 if (presetStream) preset_stream = NULL;  // we're using it up
                                 // This initial "StreamUpdateStatus()" syncs the stream according to our sync policy
                                 NormStreamObject* stream = static_cast<NormStreamObject*>(obj);
-                                if (SYNC_CURRENT == sync_policy)
+                                if ((SYNC_CURRENT == sync_policy) ||
+                                    (SYNC_REPAIR == sync_policy))
                                 {
                                     // Just "sync" to first received blockId 
                                     stream->StreamUpdateStatus(blockId);
@@ -1901,20 +1902,27 @@ void NormSenderNode::HandleObjectMessage(const NormObjectMsg& msg)
 
 bool NormSenderNode::SyncTest(const NormObjectMsg& msg) const
 {
+    // SYNC_CURRENT and SYNC_REPAIR both establish the observed object as the
+    // receiver's current point. Restrict DATA synchronization to block zero
+    // so that a later repair block cannot establish a partial object.
+    bool currentCandidate = msg.FlagIsSet(NormObjectMsg::FLAG_STREAM) ||
+                            (NormMsg::INFO == msg.GetType());
+    if (!currentCandidate && (NormMsg::DATA == msg.GetType()))
+    {
+        const NormDataMsg& data = static_cast<const NormDataMsg&>(msg);
+        currentCandidate =
+            (NormBlockId(0) == data.GetFecBlockId(fti_data.GetFecFieldSize()));
+    }
+
     switch (sync_policy)
     {
         case SYNC_CURRENT:  // default, more conservative "sync policy"
         case SYNC_STREAM:
-	    {
-            // Allow sync on stream at any time
-            bool result = msg.FlagIsSet(NormObjectMsg::FLAG_STREAM);
-            // Allow sync on INFO or block zero DATA message 
-            result = result || (NormMsg::INFO == msg.GetType()) ? 
-                                    true : (NormBlockId(0) == ((const NormDataMsg&)msg).GetFecBlockId(fti_data.GetFecFieldSize()));
-            // Never sync on repair messages
-            result = result && !msg.FlagIsSet(NormObjectMsg::FLAG_REPAIR);
-            return result;
-	    }
+            return currentCandidate &&
+                   !msg.FlagIsSet(NormObjectMsg::FLAG_REPAIR);
+
+        case SYNC_REPAIR:
+            return currentCandidate;
 
         case SYNC_ALL:     // sync on anything
             return true;
@@ -2010,6 +2018,7 @@ void NormSenderNode::Sync(NormObjectId objectId)
         {
             case SYNC_CURRENT:  // this is the usual default
             case SYNC_STREAM:
+            case SYNC_REPAIR:
                 sync_id = next_id = max_pending_object = objectId;
                 break;
             case SYNC_ALL:  // gratuitously sync for anything in our "range"

src/common/normObject.cpp

@@ -3119,6 +3119,7 @@ bool NormStreamObject::StreamUpdateStatus(NormBlockId blockId)
             read_offset = 0;
             sender->DecrementResyncCount();  // correction since stream sync here will falsely increment
             if ((NormSenderNode::SYNC_CURRENT != sender->GetSyncPolicy()) &&
+                (NormSenderNode::SYNC_REPAIR != sender->GetSyncPolicy()) &&
                 (0 != blockId.GetValue()))
             {
                 stream_broken = true;

src/dotnet/design/Mil/Navy/Nrl/Norm/Enums/NormSyncPolicy.puml

@@ -3,5 +3,6 @@ enum NormSyncPolicy {
     NORM_SYNC_CURRENT,
     NORM_SYNC_STREAM,
     NORM_SYNC_ALL,
+    NORM_SYNC_REPAIR,
 }
 @enduml

src/dotnet/src/Mil.Navy.Nrl.Norm/Enums/NormSyncPolicy.cs

@@ -17,6 +17,11 @@ public enum NormSyncPolicy
         /// Attempt recovery and reliable reception of all objects
         /// held in sender transmit object cache and newer objects.
         /// </summary>
-        NORM_SYNC_ALL
+        NORM_SYNC_ALL,
+        /// <summary>
+        /// Accept repair traffic for the observed current object without
+        /// requesting older objects from the sender cache.
+        /// </summary>
+        NORM_SYNC_REPAIR
     }
 }

src/dotnet/tests/Mil.Navy.Nrl.Norm.IntegrationTests/NormSessionTests.cs

@@ -1915,6 +1915,13 @@ public void SetsDefaultSyncPolicy_NORM_SYNC_ALL()
             _normSession.SetDefaultSyncPolicy(syncPolicy);
         }
 
+        [Fact]
+        public void SetsDefaultSyncPolicy_NORM_SYNC_REPAIR()
+        {
+            var syncPolicy = NormSyncPolicy.NORM_SYNC_REPAIR;
+            _normSession.SetDefaultSyncPolicy(syncPolicy);
+        }
+
         [Fact]
         public void SetsDefaultNackingMode_NORM_NACK_NONE()
         {
@@ -2807,4 +2814,4 @@ public void NormStreamSetsAutoFlush_FLUSH_ACTIVE()
             }
         }
     }
-}
+}

src/java/src/mil/navy/nrl/norm/enums/NormSyncPolicy.java

@@ -12,5 +12,6 @@
 public enum NormSyncPolicy {
   NORM_SYNC_CURRENT,
   NORM_SYNC_STREAM,
-  NORM_SYNC_ALL;
+  NORM_SYNC_ALL,
+  NORM_SYNC_REPAIR;
 }

src/pynorm/constants.py

@@ -75,7 +75,8 @@ class SyncPolicy(enum.Enum):
     '''
     CURRENT = 0
     STREAM  = 1
-    ALL     = 2     
+    ALL     = 2
+    REPAIR  = 3
 
 class TrackingStatus(enum.Enum):    #enum NormTrackingStatus
     NONE      = 0 

src/rust/norm/src/types.rs

@@ -242,6 +242,8 @@ pub enum SyncPolicy {
     Stream = NormSyncPolicy_NORM_SYNC_STREAM as u32,
     /// Sync to all data, old and new
     All = NormSyncPolicy_NORM_SYNC_ALL as u32,
+    /// Accept repair traffic for the observed current object only
+    Repair = NormSyncPolicy_NORM_SYNC_REPAIR as u32,
 }
 
 impl From<NormSyncPolicy> for SyncPolicy {
@@ -250,6 +252,7 @@ impl From<NormSyncPolicy> for SyncPolicy {
             NormSyncPolicy_NORM_SYNC_CURRENT => SyncPolicy::Current,
             NormSyncPolicy_NORM_SYNC_STREAM => SyncPolicy::Stream,
             NormSyncPolicy_NORM_SYNC_ALL => SyncPolicy::All,
+            NormSyncPolicy_NORM_SYNC_REPAIR => SyncPolicy::Repair,
             _ => SyncPolicy::Current,
         }
     }
@@ -261,6 +264,7 @@ impl From<SyncPolicy> for NormSyncPolicy {
             SyncPolicy::Current => NormSyncPolicy_NORM_SYNC_CURRENT,
             SyncPolicy::Stream => NormSyncPolicy_NORM_SYNC_STREAM,
             SyncPolicy::All => NormSyncPolicy_NORM_SYNC_ALL,
+            SyncPolicy::Repair => NormSyncPolicy_NORM_SYNC_REPAIR,
         }
     }
 }
@@ -522,4 +526,4 @@ impl fmt::Display for EventType {
             EventType::UserTimeout => write!(f, "UserTimeout"),
         }
     }
-}
+}

tests/normRepairFlagTest.cpp

@@ -58,8 +58,30 @@ static void TestSyncPolicy(NormSession& session)
     TEST_ASSERT(sender.SyncTest(msg));
     msg.SetFlag(NormObjectMsg::FLAG_REPAIR);
     TEST_ASSERT(!sender.SyncTest(msg));
+
+    sender.SetSyncPolicy(NormSenderNode::SYNC_REPAIR);
+    TEST_ASSERT(sender.SyncTest(msg));
+
     sender.SetSyncPolicy(NormSenderNode::SYNC_ALL);
     TEST_ASSERT(sender.SyncTest(msg));
+
+    NormDataMsg laterBlock;
+    laterBlock.Init();
+    laterBlock.SetFecId(129);
+    laterBlock.SetFecPayloadId(129, 1, 0, 1, 8);
+    laterBlock.SetFlag(NormObjectMsg::FLAG_REPAIR);
+    sender.SetSyncPolicy(NormSenderNode::SYNC_REPAIR);
+    TEST_ASSERT(!sender.SyncTest(laterBlock));
+    sender.SetSyncPolicy(NormSenderNode::SYNC_ALL);
+    TEST_ASSERT(sender.SyncTest(laterBlock));
+
+    NormInfoMsg repairInfo;
+    repairInfo.Init();
+    repairInfo.SetFlag(NormObjectMsg::FLAG_REPAIR);
+    sender.SetSyncPolicy(NormSenderNode::SYNC_CURRENT);
+    TEST_ASSERT(!sender.SyncTest(repairInfo));
+    sender.SetSyncPolicy(NormSenderNode::SYNC_REPAIR);
+    TEST_ASSERT(sender.SyncTest(repairInfo));
 }
 
 static void TestObjectMessageFlags(NormSession& session)