Better delta syncrepl support, fallback to full refresh.

diff --git a/servers/slapd/syncrepl.c b/servers/slapd/syncrepl.c
index acb0f60261a74a7f740598d5ed195a6c389df4e5..267d14e2812a4834e84b39eed5e6229b7bd9c164 100644 (file)
--- a/servers/slapd/syncrepl.c
+++ b/servers/slapd/syncrepl.c
@@ -41,14 +41,19 @@ struct nonpresent_entry {
 #define        SYNCDATA_ACCESSLOG      1       /* entries are accesslog format */
 #define        SYNCDATA_CHANGELOG      2       /* entries are changelog format */
 
+#define        SYNCLOG_LOGGING         0       /* doing a log-based update */
+#define        SYNCLOG_FALLBACK        1       /* doing a full refresh */
+
 typedef struct syncinfo_s {
        struct slap_backend_db *si_be;
        struct re_s                     *si_re;
        long                            si_rid;
        struct berval           si_provideruri;
        slap_bindconf           si_bindconf;
-       struct berval           si_filterstr;
        struct berval           si_base;
+       struct berval           si_logbase;
+       struct berval           si_filterstr;
+       struct berval           si_logfilterstr;
        int                                     si_scope;
        int                                     si_attrsonly;
        char                            *si_anfile;
@@ -71,6 +76,7 @@ typedef struct syncinfo_s {
        int                                     si_refreshDelete;
        int                                     si_refreshPresent;
        int                                     si_syncdata;
+       int                                     si_logstate;
        Avlnode                         *si_presentlist;
        LDAP                            *si_ld;
        LDAP_LIST_HEAD(np, nonpresent_entry) si_nonpresentlist;
@@ -266,6 +272,8 @@ init_syncrepl(syncinfo_t *si)
        si->si_exattrs = exattrs;       
 }
 
+static char *allattrs[] = {"*", "+", NULL};
+
 static int
 ldap_sync_search(
        syncinfo_t *si,
@@ -277,18 +285,53 @@ ldap_sync_search(
        struct timeval timeout;
        ber_int_t       msgid;
        int rc;
+       int rhint;
+       char *base;
+       char **attrs;
+       char *filter;
+       int attrsonly;
+       int stype;
+       int scope;
 
        /* setup LDAP SYNC control */
        ber_init2( ber, NULL, LBER_USE_DER );
        ber_set_option( ber, LBER_OPT_BER_MEMCTX, &ctx );
 
+       /* If we're using a log but we have no state, then fallback to
+        * normal mode for a full refresh.
+        */
+       if ( si->si_syncdata && BER_BVISEMPTY( &si->si_syncCookie.ctxcsn ))
+               si->si_logstate = SYNCLOG_FALLBACK;
+
+       /* Use the log parameters if we're in log mode */
+       if ( si->si_syncdata && si->si_logstate == SYNCLOG_LOGGING ) {
+               rhint = 0;
+               base = si->si_logbase.bv_val;
+               filter = si->si_logfilterstr.bv_val;
+               attrs = allattrs;
+               attrsonly = 0;
+               scope = LDAP_SCOPE_ONELEVEL;
+       } else {
+               rhint = 1;
+               base = si->si_base.bv_val;
+               filter = si->si_filterstr.bv_val;
+               attrs = si->si_attrs;
+               attrsonly = si->si_attrsonly;
+               scope = si->si_scope;
+       }
+       if ( si->si_syncdata && si->si_logstate == SYNCLOG_FALLBACK ) {
+               stype = LDAP_SYNC_REFRESH_ONLY;
+       } else {
+               stype = si->si_type;
+       }
+
        if ( !BER_BVISNULL( &si->si_syncCookie.octet_str ) )
        {
                ber_printf( ber, "{eOb}",
-                       abs(si->si_type), &si->si_syncCookie.octet_str, 1 );
+                       abs(stype), &si->si_syncCookie.octet_str, rhint );
        } else {
                ber_printf( ber, "{eb}",
-                       abs(si->si_type), 1 );
+                       abs(stype), rhint );
        }
 
        if ( (rc = ber_flatten2( ber, &c[0].ldctl_value, 0 )) == LBER_ERROR ) {
@@ -313,8 +356,7 @@ ldap_sync_search(
        timeout.tv_sec = si->si_tlimit;
        timeout.tv_usec = 0;
 
-       rc = ldap_search_ext( si->si_ld, si->si_base.bv_val, si->si_scope,
-               si->si_filterstr.bv_val, si->si_attrs, si->si_attrsonly,
+       rc = ldap_search_ext( si->si_ld, base, scope, filter, attrs, attrsonly,
                ctrls, NULL, si->si_tlimit > 0 ? &timeout : NULL,
                si->si_slimit, &msgid );
        ber_free_buf( ber );
@@ -621,7 +663,7 @@ do_syncrep2(
                                                slap_parse_sync_cookie( &syncCookie, NULL );
                                        }
                                }
-                               if ( si->si_syncdata ) {
+                               if ( si->si_syncdata && si->si_logstate == SYNCLOG_LOGGING ) {
                                        entry = NULL;
                                        modlist = NULL;
                                        if ( syncrepl_message_to_op( si, op, msg ) == LDAP_SUCCESS &&
@@ -657,6 +699,13 @@ do_syncrep2(
                                        "do_syncrep2: LDAP_RES_SEARCH_RESULT\n", 0, 0, 0 );
                                ldap_parse_result( si->si_ld, msg, &err, NULL, NULL, NULL,
                                        &rctrls, 0 );
+                               if ( err == LDAP_SYNC_REFRESH_REQUIRED ) {
+                                       if ( si->si_logstate == SYNCLOG_LOGGING ) {
+                                               si->si_logstate = SYNCLOG_FALLBACK;
+                                       }
+                                       rc = err;
+                                       break;
+                               }
                                if ( rctrls ) {
                                        rctrlp = *rctrls;
                                        ber_init2( ber, &rctrlp->ldctl_value, LBER_USE_DER );
@@ -712,7 +761,13 @@ do_syncrep2(
                                                si->si_presentlist = NULL;
                                        }
                                }
-                               rc = -2;
+                               if ( err == LDAP_SUCCESS
+                                       && si->si_logstate == SYNCLOG_FALLBACK ) {
+                                       si->si_logstate = SYNCLOG_LOGGING;
+                                       rc = LDAP_SYNC_REFRESH_REQUIRED;
+                               } else {
+                                       rc = -2;
+                               }
                                goto done;
                                break;
 
@@ -885,7 +940,7 @@ done:
 
        if ( res ) ldap_msgfree( res );
 
-       if ( rc && si->si_ld ) {
+       if ( rc && rc != LDAP_SYNC_REFRESH_REQUIRED && si->si_ld ) {
                ldap_unbind_ext( si->si_ld, NULL, NULL );
                si->si_ld = NULL;
        }
@@ -951,6 +1006,7 @@ do_syncrepl(
 
        /* Establish session, do search */
        if ( !si->si_ld ) {
+reload:
                first = 1;
                si->si_refreshDelete = 0;
                si->si_refreshPresent = 0;
@@ -962,6 +1018,8 @@ do_syncrepl(
                ldap_get_option( si->si_ld, LDAP_OPT_DESC, &s );
 
                rc = do_syncrep2( op, si );
+               if ( rc == LDAP_SYNC_REFRESH_REQUIRED ) 
+                       goto reload;
 
                if ( abs(si->si_type) == LDAP_SYNC_REFRESH_AND_PERSIST ) {
                        /* If we succeeded, enable the connection for further listening.
@@ -1222,6 +1280,9 @@ syncrepl_message_to_op(
                                deleteOldRdn = 1;
                } else if ( !ber_bvstrcasecmp( &bv, &ls->ls_newSup )) {
                        sup = bvals[0];
+               } else if ( !ber_bvstrcasecmp( &bv,
+                       &slap_schema.si_ad_entryCSN->ad_cname )) {
+                       slap_queue_csn( op, bvals );
                }
                ch_free( bvals );
        }
@@ -1285,6 +1346,7 @@ syncrepl_message_to_op(
                break;
        }
 done:
+       slap_graduate_commit_csn( op );
        if ( modlist )
                slap_mods_free( modlist, op->o_tag != LDAP_REQ_ADD );
        if ( !BER_BVISNULL( &rdn )) {
@@ -2540,6 +2602,8 @@ syncinfo_free( syncinfo_t *sie )
 #define SYNCDATASTR            "syncdata"
 
 /* FIXME: undocumented */
+#define        LOGBASESTR      "logbase"
+#define LOGFILTERSTR   "logfilter"
 #define OLDAUTHCSTR            "bindprincipal"
 #define EXATTRSSTR             "exattrs"
 #define MANAGEDSAITSTR         "manageDSAit"
@@ -2632,6 +2696,13 @@ parse_syncrepl_line(
                        if ( si->si_filterstr.bv_val )
                                ch_free( si->si_filterstr.bv_val );
                        ber_str2bv( val, 0, 1, &si->si_filterstr );
+               } else if ( !strncasecmp( cargv[ i ], LOGFILTERSTR "=",
+                                       STRLENOF( LOGFILTERSTR "=" ) ) )
+               {
+                       val = cargv[ i ] + STRLENOF( LOGFILTERSTR "=" );
+                       if ( si->si_logfilterstr.bv_val )
+                               ch_free( si->si_logfilterstr.bv_val );
+                       ber_str2bv( val, 0, 1, &si->si_logfilterstr );
                } else if ( !strncasecmp( cargv[ i ], SEARCHBASESTR "=",
                                        STRLENOF( SEARCHBASESTR "=" ) ) )
                {
@@ -2649,6 +2720,23 @@ parse_syncrepl_line(
                                        val, rc, ldap_err2string( rc ) );
                                return -1;
                        }
+               } else if ( !strncasecmp( cargv[ i ], LOGBASESTR "=",
+                                       STRLENOF( LOGBASESTR "=" ) ) )
+               {
+                       struct berval   bv;
+                       int             rc;
+
+                       val = cargv[ i ] + STRLENOF( LOGBASESTR "=" );
+                       if ( si->si_logbase.bv_val ) {
+                               ch_free( si->si_logbase.bv_val );
+                       }
+                       ber_str2bv( val, 0, 0, &bv );
+                       rc = dnNormalize( 0, NULL, NULL, &bv, &si->si_logbase, NULL );
+                       if ( rc != LDAP_SUCCESS ) {
+                               fprintf( stderr, "Invalid logbase DN \"%s\": %d (%s)\n",
+                                       val, rc, ldap_err2string( rc ) );
+                               return -1;
+                       }
                } else if ( !strncasecmp( cargv[ i ], SCOPESTR "=",
                                        STRLENOF( SCOPESTR "=" ) ) )
                {
@@ -2980,6 +3068,16 @@ syncrepl_unparse( syncinfo_t *si, struct berval *bv )
                ptr = lutil_strcopy( ptr, si->si_base.bv_val );
                *ptr++ = '"';
        }
+       if ( !BER_BVISEMPTY( &si->si_logfilterstr )) {
+               ptr = lutil_strcopy( ptr, " " LOGFILTERSTR "=\"" );
+               ptr = lutil_strcopy( ptr, si->si_logfilterstr.bv_val );
+               *ptr++ = '"';
+       }
+       if ( !BER_BVISNULL( &si->si_logbase )) {
+               ptr = lutil_strcopy( ptr, " " LOGBASESTR "=\"" );
+               ptr = lutil_strcopy( ptr, si->si_logbase.bv_val );
+               *ptr++ = '"';
+       }
        for (i=0; !BER_BVISNULL(&scopes[i].key);i++) {
                if ( si->si_scope == scopes[i].val ) {
                        ptr = lutil_strcopy( ptr, " " SCOPESTR "=" );