Better delta syncrepl support, fallback to full refresh.

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,
+	rc = ldap_search_ext( si->si_ld, base, scope, filter, attrs, attrsonly,
-		si->si_filterstr.bv_val, si->si_attrs, si->si_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;
 					}
 				}
+				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 "=" );