Two-phase HDB commit via iprop log, + GC for log

We used to update the iprop log and HDB in different orders depending on the kadm5 operation, which then led to various race conditions. The iprop log now functions as a two-phase commit (with roll forward) log for HDB changes. The log is auto-truncated, keeping the latest entries that fit in a configurable maximum number of bytes (defaults to 50MB). See the log-max-size parameter description in krb5.conf(5). The iprop log format and the protocol remain backwards-compatible with earlier versions of Heimdal. This is NOT a flag-day; there is NO need to update all the slaves at once with the master, though it is advisable in general. Rolling upgrades and downgrades should work. The sequence of updates is now (with HDB and log open and locked): a) check that the HDB operation will succeed if attempted, b) append to iprop log and fsync() it, c) write to HDB (which should fsync()), d) mark last log record committed (no fsync in this case). Every kadm5 write operation recover transactions not yet confirmed as committed, thus there can be at most one unconfirmed commit on a master KDC. Reads via kadm5_get_principal() also attempt to lock the log, and if successful, recover unconfirmed transactions; readers must have write access and must win any race to lock the iprop log. The ipropd-master daemon also attempts to recover unconfirmed transactions when idle. The log now starts with a nop record whose payload records the offset of the logical end of the log: the end of the last confirmed committed transaction. This is kown as the "uber record". Its purpose is two-fold: act as the confirmation of committed transactions, and provide an O(1) method of finding the end of the log (i.e., without having to traverse the entire log front to back). Two-phase commit makes all kadm5 writes single-operation atomic transactions (though some kadm5 operations, such as renames of principals, and changes to principals' aliases, use multiple low-level HDB write operations, but still all in one transaction). One can still hold a lock on the HDB across many operations (e.g., by using the lock command in a kadmin -l or calling kadm5_lock()) in order to push multiple transactions in sequence, but this sequence will not be atomic if the process or host crashes in the middle. As before, HDB writes which do not go through the kadm5 API are excluded from all of this, but there should be no such writes. Lastly, the iprop-log(1) command is enhanced as follows: - The dump, last-version, truncate, and replay sub-commands now have an option to not lock the log. This is useful for inspecting a running system's log file, especially on slave KDCs. - The dump, last-version, truncate, and replay sub-commands now take an optional iprop log file positional argument, so that they may be used to inspect log files other than the running system's configured/default log file. Extensive code review and some re-writing for clarity by Viktor Dukhovni.
2015-05-29 15:53:40 -05:00
parent d774aeda38
commit 20df2c8706
33 changed files with 3274 additions and 804 deletions
--- a/kadmin/init.c
+++ b/kadmin/init.c
@@ -157,14 +157,11 @@ init(struct init_options *opt, int argc, char **argv)
 	krb5_warn(context, ret, "hdb_open");
 	return 0;
    }
-    db->hdb_close(context, db);
-    ret = kadm5_log_init(kadm_handle);
-    if (ret)
-        krb5_err(context, 1, ret, "Failed iprop log initialization");
-    ret = kadm5_log_nop(kadm_handle);
+    ret = kadm5_log_reinit(kadm_handle);
    if (ret)
        krb5_err(context, 1, ret, "Failed iprop log initialization");
    kadm5_log_end(kadm_handle);
+    db->hdb_close(context, db);
    for(i = 0; i < argc; i++){
 	krb5_principal princ;
 	const char *realm = argv[i];
--- a/kadmin/load.c
+++ b/kadmin/load.c
@@ -377,10 +377,17 @@ doit(const char *filename, int mergep)
 	krb5_warn(context, errno, "fopen(%s)", filename);
 	return 1;
    }
-    ret = kadm5_log_truncate (kadm_handle);
+    /*
+     * We don't have a version number in the dump, so we don't know
+     * which iprop log entries to keep, if any.  We throw the log away.
+     *
+     * We could merge the ipropd-master/slave dump/load here as an
+     * option, in which case we would first load the dump.
+     */
+    ret = kadm5_log_reinit(kadm_handle);
    if (ret) {
 	fclose (f);
-	krb5_warn(context, ret, "kadm5_log_truncate");
+	krb5_warn(context, ret, "kadm5_log_reinit");
 	return 1;
    }

@@ -535,6 +542,7 @@ doit(const char *filename, int mergep)
 	    break;
 	}
    }
+    (void) kadm5_log_end(kadm_handle);
    db->hdb_close(context, db);
    fclose(f);
    return ret != 0;
--- a/lib/hdb/common.c
+++ b/lib/hdb/common.c
@@ -321,6 +321,22 @@ _hdb_store(krb5_context context, HDB *db, unsigned flags, hdb_entry_ex *entry)
    if (code)
 	return code;

+    if ((flags & HDB_F_PRECHECK) && (flags & HDB_F_REPLACE))
+        return 0;
+
+    if ((flags & HDB_F_PRECHECK)) {
+        code = hdb_principal2key(context, entry->entry.principal, &key);
+        if (code)
+            return code;
+        code = db->hdb__get(context, db, key, &value);
+        krb5_data_free(&key);
+        if (code == 0)
+            krb5_data_free(&value);
+        if (code == HDB_ERR_NOENTRY)
+            return 0;
+        return code ? code : HDB_ERR_EXISTS;
+    }
+
    if(entry->entry.generation == NULL) {
 	struct timeval t;
 	entry->entry.generation = malloc(sizeof(*entry->entry.generation));
@@ -360,13 +376,32 @@ _hdb_store(krb5_context context, HDB *db, unsigned flags, hdb_entry_ex *entry)
 }

 krb5_error_code
-_hdb_remove(krb5_context context, HDB *db, krb5_const_principal principal)
+_hdb_remove(krb5_context context, HDB *db,
+            unsigned flags, krb5_const_principal principal)
 {
-    krb5_data key;
+    krb5_data key, value;
    int code;

    hdb_principal2key(context, principal, &key);

+    if ((flags & HDB_F_PRECHECK)) {
+        /*
+         * We don't check that we can delete the aliases because we
+         * assume that the DB is consistent.  If we did check for alias
+         * consistency we'd also have to provide a way to fsck the DB,
+         * otherwise admins would have no way to recover -- papering
+         * over this here is less work, but we really ought to provide
+         * an HDB fsck.
+         */
+        code = db->hdb__get(context, db, key, &value);
+        krb5_data_free(&key);
+        if (code == 0) {
+            krb5_data_free(&value);
+            return 0;
+        }
+        return code;
+    }
+
    code = hdb_remove_aliases(context, db, &key);
    if (code) {
 	krb5_data_free(&key);
--- a/lib/hdb/db.c
+++ b/lib/hdb/db.c
@@ -211,6 +211,7 @@ DB__put(krb5_context context, HDB *db, int replace,
    k.size = key.length;
    v.data = value.data;
    v.size = value.length;
+    krb5_clear_error_message(context);
    code = (*d->put)(d, &k, &v, replace ? 0 : R_NOOVERWRITE);
    if(code < 0) {
 	code = errno;
@@ -219,9 +220,15 @@ DB__put(krb5_context context, HDB *db, int replace,
 	return code;
    }
    if(code == 1) {
-	krb5_clear_error_message(context);
 	return HDB_ERR_EXISTS;
    }
+    code = (*d->sync)(d, 0);
+    if (code == -1) {
+	code = errno;
+	krb5_set_error_message(context, code, "Database %s put sync error: %s",
+			       db->hdb_name, strerror(code));
+	return code;
+    }
    return 0;
 }

@@ -233,15 +240,23 @@ DB__del(krb5_context context, HDB *db, krb5_data key)
    krb5_error_code code;
    k.data = key.data;
    k.size = key.length;
+    krb5_clear_error_message(context);
    code = (*d->del)(d, &k, 0);
-    if(code == 1) {
+    if (code == 1)
+        return HDB_ERR_NOENTRY;
+    if (code < 0) {
 	code = errno;
-	krb5_set_error_message(context, code, "Database %s put error: %s",
+	krb5_set_error_message(context, code, "Database %s del error: %s",
+			       db->hdb_name, strerror(code));
+	return code;
+    }
+    code = (*d->sync)(d, 0);
+    if (code == -1) {
+	code = errno;
+	krb5_set_error_message(context, code, "Database %s del sync error: %s",
 			       db->hdb_name, strerror(code));
 	return code;
    }
-    if(code < 0)
-	return errno;
    return 0;
 }

--- a/lib/hdb/db3.c
+++ b/lib/hdb/db3.c
@@ -235,8 +235,46 @@ DB__put(krb5_context context, HDB *db, int replace,
    code = (*d->put)(d, NULL, &k, &v, replace ? 0 : DB_NOOVERWRITE);
    if(code == DB_KEYEXIST)
 	return HDB_ERR_EXISTS;
-    if(code)
-	return errno;
+    if (code) {
+        /*
+         * Berkeley DB 3 and up have a terrible error reporting
+         * interface...
+         *
+         * DB->err() doesn't output a string.
+         * DB->set_errcall()'s callback function doesn't have a void *
+         * argument that can be used to place the error somewhere.
+         *
+         * The only thing we could do is fopen()/fdopen() a file, set it
+         * with DB->set_errfile(), then call DB->err(), then read the
+         * message from the file, unset it with DB->set_errfile(), close
+         * it and delete it.  That's a lot of work... so we don't do it.
+         */
+        if (code == EACCES || code == ENOSPC || code == EINVAL) {
+            krb5_set_error_message(context, code,
+                                   "Database %s put error: %s",
+                                   db->hdb_name, strerror(code));
+        } else {
+            code = HDB_ERR_UK_SERROR;
+            krb5_set_error_message(context, code,
+                                   "Database %s put error: unknown (%d)",
+                                   db->hdb_name, code);
+        }
+	return code;
+    }
+    code = (*d->sync)(d, 0);
+    if (code) {
+        if (code == EACCES || code == ENOSPC || code == EINVAL) {
+            krb5_set_error_message(context, code,
+                                   "Database %s put sync error: %s",
+                                   db->hdb_name, strerror(code));
+        } else {
+            code = HDB_ERR_UK_SERROR;
+            krb5_set_error_message(context, code,
+                                   "Database %s put sync error: unknown (%d)",
+                                   db->hdb_name, code);
+        }
+        return code;
+    }
    return 0;
 }

@@ -253,8 +291,33 @@ DB__del(krb5_context context, HDB *db, krb5_data key)
    code = (*d->del)(d, NULL, &k, 0);
    if(code == DB_NOTFOUND)
 	return HDB_ERR_NOENTRY;
-    if(code)
+    if (code) {
+        if (code == EACCES || code == ENOSPC || code == EINVAL) {
+            krb5_set_error_message(context, code,
+                                   "Database %s del error: %s",
+                                   db->hdb_name, strerror(code));
+        } else {
+            code = HDB_ERR_UK_SERROR;
+            krb5_set_error_message(context, code,
+                                   "Database %s del error: unknown (%d)",
+                                   db->hdb_name, code);
+        }
 	return code;
+    }
+    code = (*d->sync)(d, 0);
+    if (code) {
+        if (code == EACCES || code == ENOSPC || code == EINVAL) {
+            krb5_set_error_message(context, code,
+                                   "Database %s del sync error: %s",
+                                   db->hdb_name, strerror(code));
+        } else {
+            code = HDB_ERR_UK_SERROR;
+            krb5_set_error_message(context, code,
+                                   "Database %s del sync error: unknown (%d)",
+                                   db->hdb_name, code);
+        }
+        return code;
+    }
    return 0;
 }

--- a/lib/hdb/hdb-ldap.c
+++ b/lib/hdb/hdb-ldap.c
@@ -1735,6 +1735,9 @@ LDAP_store(krb5_context context, HDB * db, unsigned flags,
    LDAPMessage *msg = NULL, *e = NULL;
    char *dn = NULL, *name = NULL;

+    if ((flags & HDB_F_PRECHECK))
+        return 0; /* we can't guarantee whether we'll be able to perform it */
+
    ret = LDAP_principal2message(context, db, entry->entry.principal, &msg);
    if (ret == 0)
 	e = ldap_first_entry(HDB2LDAP(db), msg);
@@ -1806,13 +1809,17 @@ LDAP_store(krb5_context context, HDB * db, unsigned flags,
 }

 static krb5_error_code
-LDAP_remove(krb5_context context, HDB *db, krb5_const_principal principal)
+LDAP_remove(krb5_context context, HDB *db,
+            unsigned flags, krb5_const_principal principal)
 {
    krb5_error_code ret;
    LDAPMessage *msg, *e;
    char *dn = NULL;
    int rc, limit = LDAP_NO_LIMIT;

+    if ((flags & HDB_F_PRECHECK))
+        return 0; /* we can't guarantee whether we'll be able to perform it */
+
    ret = LDAP_principal2message(context, db, principal, &msg);
    if (ret)
 	goto out;
--- a/lib/hdb/hdb-mitdb.c
+++ b/lib/hdb/hdb-mitdb.c
@@ -950,8 +950,24 @@ mdb_store(krb5_context context, HDB *db, unsigned flags, hdb_entry_ex *entry)
    krb5_data line = { 0, 0 };
    krb5_data kdb_ent = { 0, 0 };
    krb5_data key = { 0, 0 };
+    krb5_data value = { 0, 0 };
    ssize_t sz;

+    if ((flags & HDB_F_PRECHECK) && (flags & HDB_F_REPLACE))
+        return 0;
+
+    if ((flags & HDB_F_PRECHECK)) {
+        ret = mdb_principal2key(context, entry->entry.principal, &key);
+        if (ret) return ret;
+        code = db->hdb__get(context, db, key, &value);
+        krb5_data_free(&key);
+        if (code == 0)
+            krb5_data_free(&value);
+        if (code == HDB_ERR_NOENTRY)
+            return 0;
+        return code ? code : HDB_ERR_EXISTS;
+    }
+
    sp = krb5_storage_emem();
    if (!sp) return ENOMEM;
    ret = _hdb_set_master_key_usage(context, db, 0); /* MIT KDB uses KU 0 */
@@ -989,11 +1005,22 @@ out:
 }

 static krb5_error_code
-mdb_remove(krb5_context context, HDB *db, krb5_const_principal principal)
+mdb_remove(krb5_context context, HDB *db,
+           unsigned flags, krb5_const_principal principal)
 {
    krb5_error_code code;
    krb5_data key;

+    if ((flags & HDB_F_PRECHECK)) {
+        code = db->hdb__get(context, db, key, &value);
+        krb5_data_free(&key);
+        if (code == 0) {
+            krb5_data_free(&value);
+            return 0;
+        }
+        return code;
+    }
+
    mdb_principal2key(context, principal, &key);
    code = db->hdb__del(context, db, key);
    krb5_data_free(&key);
--- a/lib/hdb/hdb-sqlite.c
+++ b/lib/hdb/hdb-sqlite.c
@@ -672,24 +672,24 @@ hdb_sqlite_store(krb5_context context, HDB *db, unsigned flags,
            goto rollback;
    }

-    ret = 0;
-
 commit:
-
    krb5_data_free(&value);
-
    sqlite3_clear_bindings(get_ids);
    sqlite3_reset(get_ids);

+    if ((flags & HDB_F_PRECHECK)) {
+        (void) hdb_sqlite_exec_stmt(context, hsdb, "ROLLBACK", ret);
+        return 0;
+    }
+
    ret = hdb_sqlite_exec_stmt(context, hsdb, "COMMIT", EINVAL);
    if(ret != SQLITE_OK)
 	krb5_warnx(context, "hdb-sqlite: COMMIT problem: %d: %s",
 		   ret, sqlite3_errmsg(hsdb->db));

-    return ret;
+    return ret == SQLITE_OK ? 0 : HDB_ERR_UK_SERROR;

 rollback:
-
    krb5_warnx(context, "hdb-sqlite: store rollback problem: %d: %s",
 	       ret, sqlite3_errmsg(hsdb->db));

@@ -865,27 +865,63 @@ hdb_sqlite_rename(krb5_context context, HDB *db, const char *new_name)
 */
 static krb5_error_code
 hdb_sqlite_remove(krb5_context context, HDB *db,
-                  krb5_const_principal principal)
+                  unsigned flags, krb5_const_principal principal)
 {
    krb5_error_code ret;
    hdb_sqlite_db *hsdb = (hdb_sqlite_db*)(db->hdb_db);
+    sqlite3_stmt *get_ids = hsdb->get_ids;
    sqlite3_stmt *rm = hsdb->remove;

    bind_principal(context, principal, rm, 1);

+    ret = hdb_sqlite_exec_stmt(context, hsdb,
+                               "BEGIN IMMEDIATE TRANSACTION", EINVAL);
+    if (ret != SQLITE_OK) {
+        (void) hdb_sqlite_exec_stmt(context, hsdb, "ROLLBACK", EINVAL);
+	ret = HDB_ERR_UK_SERROR;
+        krb5_set_error_message(context, ret,
+			       "SQLite BEGIN TRANSACTION failed: %s",
+			       sqlite3_errmsg(hsdb->db));
+        return ret;
+    }
+
+    if ((flags & HDB_F_PRECHECK)) {
+        ret = bind_principal(context, principal, get_ids, 1);
+        if (ret)
+            return ret;
+
+        ret = hdb_sqlite_step(context, hsdb->db, get_ids);
+        sqlite3_clear_bindings(get_ids);
+        sqlite3_reset(get_ids);
+        if (ret == SQLITE_DONE) {
+            (void) hdb_sqlite_exec_stmt(context, hsdb, "ROLLBACK", EINVAL);
+            return HDB_ERR_NOENTRY;
+        }
+    }
+
    ret = hdb_sqlite_step(context, hsdb->db, rm);
+    sqlite3_clear_bindings(rm);
+    sqlite3_reset(rm);
    if (ret != SQLITE_DONE) {
-	ret = EINVAL;
+        (void) hdb_sqlite_exec_stmt(context, hsdb, "ROLLBACK", ret);
+	ret = HDB_ERR_UK_SERROR;
        krb5_set_error_message(context, ret,
                              "sqlite remove failed: %d",
                              ret);
-    } else
-        ret = 0;
+        return ret;
+    }

-    sqlite3_clear_bindings(rm);
-    sqlite3_reset(rm);
+    if ((flags & HDB_F_PRECHECK)) {
+        (void) hdb_sqlite_exec_stmt(context, hsdb, "ROLLBACK", EINVAL);
+        return 0;
+    }

-    return ret;
+    ret = hdb_sqlite_exec_stmt(context, hsdb, "COMMIT", HDB_ERR_UK_SERROR);
+    if (ret != SQLITE_OK)
+	krb5_warnx(context, "hdb-sqlite: COMMIT problem: %ld: %s",
+		   (long)HDB_ERR_UK_SERROR, sqlite3_errmsg(hsdb->db));
+
+    return 0;
 }

 /**
--- a/lib/hdb/hdb.h
+++ b/lib/hdb/hdb.h
@@ -65,6 +65,7 @@ enum hdb_lockop{ HDB_RLOCK, HDB_WLOCK };
 #define HDB_F_ALL_KVNOS		2048	/* we want all the keys, live or not */
 #define HDB_F_FOR_AS_REQ	4096	/* fetch is for a AS REQ */
 #define HDB_F_FOR_TGS_REQ	8192	/* fetch is for a TGS REQ */
+#define HDB_F_PRECHECK		16384	/* check that the operation would succeed */

 /* hdb_capability_flags */
 #define HDB_CAP_F_HANDLE_ENTERPRISE_PRINCIPAL 1
@@ -154,7 +155,7 @@ typedef struct HDB {
     * Remove an entry from the database.
     */
    krb5_error_code (*hdb_remove)(krb5_context, struct HDB*,
-				  krb5_const_principal);
+				  unsigned, krb5_const_principal);
    /**
     * As part of iteration, fetch one entry
     */
@@ -186,25 +187,33 @@ typedef struct HDB {
    /**
     * Get an hdb_entry from a classical DB backend
     *
-     * If the database is a classical DB (ie BDB, NDBM, GDBM, etc)
-     * backend, this function will take a principal key (krb5_data)
-     * and return all data related to principal in the return
-     * krb5_data. The returned encoded entry is of type hdb_entry or
-     * hdb_entry_alias.
+     * This function takes a principal key (krb5_data) and returns all
+     * data related to principal in the return krb5_data. The returned
+     * encoded entry is of type hdb_entry or hdb_entry_alias.
     */
    krb5_error_code (*hdb__get)(krb5_context, struct HDB*,
 				krb5_data, krb5_data*);
    /**
     * Store an hdb_entry from a classical DB backend
     *
-     * Same discussion as in @ref HDB::hdb__get
+     * This function takes a principal key (krb5_data) and encoded
+     * hdb_entry or hdb_entry_alias as the data to store.
+     *
+     * For a file-based DB, this must synchronize to disk when done.
+     * This is sub-optimal for kadm5_s_rename_principal(), and for
+     * kadm5_s_modify_principal() when using principal aliases; to
+     * improve this so that only one fsync() need be done
+     * per-transaction will require HDB API extensions.
     */
    krb5_error_code (*hdb__put)(krb5_context, struct HDB*, int,
 				krb5_data, krb5_data);
    /**
     * Delete and hdb_entry from a classical DB backend
     *
-     * Same discussion as in @ref HDB::hdb__get
+     * This function takes a principal key (krb5_data) naming the record
+     * to delete.
+     *
+     * Same discussion as in @ref HDB::hdb__put
     */
    krb5_error_code (*hdb__del)(krb5_context, struct HDB*, krb5_data);
    /**
@@ -263,7 +272,7 @@ typedef struct HDB {
    krb5_error_code (*hdb_check_s4u2self)(krb5_context, struct HDB *, hdb_entry_ex *, krb5_const_principal);
 }HDB;

-#define HDB_INTERFACE_VERSION	8
+#define HDB_INTERFACE_VERSION	9

 struct hdb_method {
    int			version;
--- a/lib/kadm5/admin.h
+++ b/lib/kadm5/admin.h
@@ -48,6 +48,10 @@

 #define KADM5_STRUCT_VERSION 0

+/* For kadm5_log_get_version_fd() */
+#define LOG_VERSION_LAST -1
+#define LOG_VERSION_FIRST 1
+
 #include <krb5.h>

 #define KRB5_KDB_DISALLOW_POSTDATED	0x00000001
--- a/lib/kadm5/chpass_s.c
+++ b/lib/kadm5/chpass_s.c
@@ -58,6 +58,10 @@ change(void *server_handle,
 	    return ret;
    }

+    ret = kadm5_log_init(context);
+    if (ret)
+        goto out;
+
    ret = context->db->hdb_fetch_kvno(context->context, context->db, princ,
 				      HDB_F_DECRYPT|HDB_F_GET_ANY|HDB_F_ADMIN_DATA, 0, &ent);
    if(ret)
@@ -138,21 +142,23 @@ change(void *server_handle,
    if (ret)
 	goto out2;

-    ret = context->db->hdb_store(context->context, context->db,
-				 HDB_F_REPLACE, &ent);
-    if (ret)
-	goto out2;
-
-    kadm5_log_modify(context, &ent.entry,
-		     KADM5_ATTRIBUTES | KADM5_PRINCIPAL | KADM5_MOD_NAME |
-		     KADM5_MOD_TIME | KADM5_KEY_DATA | KADM5_KVNO |
-		     KADM5_PW_EXPIRATION | KADM5_TL_DATA);
+    /* This logs the change for iprop and writes to the HDB */
+    ret = kadm5_log_modify(context, &ent.entry,
+                           KADM5_ATTRIBUTES | KADM5_PRINCIPAL |
+                           KADM5_MOD_NAME | KADM5_MOD_TIME |
+                           KADM5_KEY_DATA | KADM5_KVNO |
+                           KADM5_PW_EXPIRATION | KADM5_TL_DATA);

 out2:
    hdb_free_entry(context->context, &ent);
 out:
-    if (!context->keep_open)
-	context->db->hdb_close(context->context, context->db);
+    (void) kadm5_log_end(context);
+    if (!context->keep_open) {
+        kadm5_ret_t ret2;
+        ret2 = context->db->hdb_close(context->context, context->db);
+        if (ret == 0 && ret2 != 0)
+            ret = ret2;
+    }
    return _kadm5_error_code(ret);
 }

@@ -208,6 +214,11 @@ kadm5_s_chpass_principal_with_key(void *server_handle,
 	if(ret)
 	    return ret;
    }
+
+    ret = kadm5_log_init(context);
+    if (ret)
+        goto out;
+
    ret = context->db->hdb_fetch_kvno(context->context, context->db, princ, 0,
 				      HDB_F_GET_ANY|HDB_F_ADMIN_DATA, &ent);
    if(ret == HDB_ERR_NOENTRY)
@@ -243,22 +254,21 @@ kadm5_s_chpass_principal_with_key(void *server_handle,
 	hdb_replace_extension(context->context, &ent.entry, &ext);
    }

-
-    ret = context->db->hdb_store(context->context, context->db,
-				 HDB_F_REPLACE, &ent);
-    if (ret)
-	goto out2;
-
-    kadm5_log_modify (context,
-		      &ent.entry,
-		      KADM5_PRINCIPAL | KADM5_MOD_NAME | KADM5_MOD_TIME |
-		      KADM5_KEY_DATA | KADM5_KVNO | KADM5_PW_EXPIRATION |
-		      KADM5_TL_DATA);
+    /* This logs the change for iprop and writes to the HDB */
+    ret = kadm5_log_modify(context, &ent.entry,
+                           KADM5_PRINCIPAL | KADM5_MOD_NAME |
+                           KADM5_MOD_TIME | KADM5_KEY_DATA | KADM5_KVNO |
+                           KADM5_PW_EXPIRATION | KADM5_TL_DATA);

 out2:
    hdb_free_entry(context->context, &ent);
 out:
-    if (!context->keep_open)
-	context->db->hdb_close(context->context, context->db);
+    (void) kadm5_log_end(context);
+    if (!context->keep_open) {
+        kadm5_ret_t ret2;
+        ret2 = context->db->hdb_close(context->context, context->db);
+        if (ret == 0 && ret2 != 0)
+            ret = ret2;
+    }
    return _kadm5_error_code(ret);
 }
--- a/lib/kadm5/create_s.c
+++ b/lib/kadm5/create_s.c
@@ -127,23 +127,38 @@ kadm5_s_create_principal_with_key(void *server_handle,
    if(ret)
 	goto out;

+    if (!context->keep_open) {
+        ret = context->db->hdb_open(context->context, context->db, O_RDWR, 0);
+        if (ret)
+            goto out;
+    }
+
+    ret = kadm5_log_init(context);
+    if (ret)
+        goto out;
+
+    ret = context->db->hdb_fetch_kvno(context->context, context->db,
+                                      princ->principal, 0, 0, &ent);
+    if (ret == 0) {
+        ret = HDB_ERR_EXISTS;
+        goto out;
+    }
+
    ret = hdb_seal_keys(context->context, context->db, &ent.entry);
    if (ret)
 	goto out;

-    if (!context->keep_open) {
-	ret = context->db->hdb_open(context->context, context->db, O_RDWR, 0);
-	if(ret)
-	    goto out;
-    }
-    ret = context->db->hdb_store(context->context, context->db, 0, &ent);
-    if (!context->keep_open)
-	context->db->hdb_close(context->context, context->db);
-    if (ret)
-	goto out;
-    kadm5_log_create (context, &ent.entry);
+    /* This logs the change for iprop and writes to the HDB */
+    ret = kadm5_log_create(context, &ent.entry);

 out:
+    (void) kadm5_log_end(context);
+    if (!context->keep_open) {
+        kadm5_ret_t ret2;
+        ret2 = context->db->hdb_close(context->context, context->db);
+        if (ret == 0 && ret2 != 0)
+            ret = ret2;
+    }
    hdb_free_entry(context->context, &ent);
    return _kadm5_error_code(ret);
 }
@@ -174,9 +189,26 @@ kadm5_s_create_principal(void *server_handle,
 			   | KADM5_AUX_ATTRIBUTES | KADM5_KEY_DATA
 			   | KADM5_POLICY_CLR | KADM5_LAST_SUCCESS
 			   | KADM5_LAST_FAILED | KADM5_FAIL_AUTH_COUNT);
-    if(ret)
+    if (ret)
 	goto out;

+    if (!context->keep_open) {
+        ret = context->db->hdb_open(context->context, context->db, O_RDWR, 0);
+        if (ret)
+            goto out;
+    }
+
+    ret = kadm5_log_init(context);
+    if (ret)
+        goto out;
+
+    ret = context->db->hdb_fetch_kvno(context->context, context->db,
+                                      princ->principal, 0, 0, &ent);
+    if (ret == 0) {
+        ret = HDB_ERR_EXISTS;
+        goto out;
+    }
+
    ent.entry.keys.len = 0;
    ent.entry.keys.val = NULL;

@@ -188,20 +220,17 @@ kadm5_s_create_principal(void *server_handle,
    if (ret)
 	goto out;

-    if (!context->keep_open) {
-	ret = context->db->hdb_open(context->context, context->db, O_RDWR, 0);
-	if(ret)
-	    goto out;
-    }
-    ret = context->db->hdb_store(context->context, context->db, 0, &ent);
-    if (!context->keep_open)
-	context->db->hdb_close(context->context, context->db);
-    if (ret)
-	goto out;
-
-    kadm5_log_create (context, &ent.entry);
+    /* This logs the change for iprop and writes to the HDB */
+    ret = kadm5_log_create(context, &ent.entry);

 out:
+    (void) kadm5_log_end(context);
+    if (!context->keep_open) {
+        kadm5_ret_t ret2;
+        ret2 = context->db->hdb_close(context->context, context->db);
+        if (ret == 0 && ret2 != 0)
+            ret = ret2;
+    }
    hdb_free_entry(context->context, &ent);
    return _kadm5_error_code(ret);
 }
--- a/lib/kadm5/delete_s.c
+++ b/lib/kadm5/delete_s.c
@@ -50,11 +50,16 @@ kadm5_s_delete_principal(void *server_handle, krb5_principal princ)
 	    return ret;
 	}
    }
+
+    ret = kadm5_log_init(context);
+    if (ret)
+        goto out;
+
    ret = context->db->hdb_fetch_kvno(context->context, context->db, princ,
 				      HDB_F_DECRYPT|HDB_F_GET_ANY|HDB_F_ADMIN_DATA, 0, &ent);
-    if(ret == HDB_ERR_NOENTRY)
+    if (ret == HDB_ERR_NOENTRY)
 	goto out;
-    if(ent.entry.flags.immutable) {
+    if (ent.entry.flags.immutable) {
 	ret = KADM5_PROTECT_PRINCIPAL;
 	goto out2;
    }
@@ -63,16 +68,18 @@ kadm5_s_delete_principal(void *server_handle, krb5_principal princ)
    if (ret)
 	goto out2;

-    ret = context->db->hdb_remove(context->context, context->db, princ);
-    if (ret)
-	goto out2;
-
-    kadm5_log_delete (context, princ);
+    /* This logs the change for iprop and writes to the HDB */
+    ret = kadm5_log_delete(context, princ);

 out2:
    hdb_free_entry(context->context, &ent);
 out:
-    if (!context->keep_open)
-	context->db->hdb_close(context->context, context->db);
+    (void) kadm5_log_end(context);
+    if (!context->keep_open) {
+        kadm5_ret_t ret2;
+        ret2 = context->db->hdb_close(context->context, context->db);
+        if (ret == 0 && ret2 != 0)
+            ret = ret2;
+    }
    return _kadm5_error_code(ret);
 }
--- a/lib/kadm5/get_s.c
+++ b/lib/kadm5/get_s.c
@@ -125,17 +125,37 @@ kadm5_s_get_principal(void *server_handle,
    kadm5_server_context *context = server_handle;
    kadm5_ret_t ret;
    hdb_entry_ex ent;
+    int hdb_is_rw = 1;

    memset(&ent, 0, sizeof(ent));

    if (!context->keep_open) {
-	ret = context->db->hdb_open(context->context, context->db, O_RDONLY, 0);
+	ret = context->db->hdb_open(context->context, context->db, O_RDWR, 0);
+        if (ret == EPERM || ret == EACCES) {
+            ret = context->db->hdb_open(context->context, context->db, O_RDONLY, 0);
+            hdb_is_rw = 0;
+        }
 	if(ret)
 	    return ret;
    }
+
+    /*
+     * Attempt to recover the log.  This will generally fail on slaves,
+     * and we can't tell if we're on a slave here.
+     *
+     * Perhaps we could set a flag in the kadm5_server_context to
+     * indicate whether a read has been done without recovering the log,
+     * in which case we could fail any subsequent writes.
+     */
+    if (hdb_is_rw)
+        (void) kadm5_log_init_nb(context);
+
    ret = context->db->hdb_fetch_kvno(context->context, context->db, princ,
 				      HDB_F_DECRYPT|HDB_F_ALL_KVNOS|
 				      HDB_F_GET_ANY|HDB_F_ADMIN_DATA, 0, &ent);
+    if (hdb_is_rw)
+        kadm5_log_end(context);
+
    if (!context->keep_open)
 	context->db->hdb_close(context->context, context->db);
    if(ret)
--- a/lib/kadm5/iprop-commands.in
+++ b/lib/kadm5/iprop-commands.in
@@ -41,6 +41,18 @@ command = {
 		help = "configuration file"
 		argument = "file"
 	}
+	option = {
+		long = "no-lock"
+		short = "n"
+		type = "flag"
+		help = "don't lock iprop log"
+	}
+	option = {
+		long = "reverse"
+		short = "R"
+		type = "flag"
+		help = "dump the log in reverse order"
+	}
 	option = {
 		long = "realm"
 		short = "r"
@@ -49,7 +61,7 @@ command = {
 	}
 	function = "iprop_dump"
 	help = "Prints the iprop transaction log in text."
-	max_args = "0"
+	max_args = "1"
 }
 command = {
 	name = "truncate"
@@ -66,9 +78,30 @@ command = {
 		type = "string"
 		help = "realm"
 	}
+	option = {
+		long = "keep-entries"
+		short = "K"
+		type = "integer"
+		help = "number of entries to keep"
+                default = "-1"
+	}
+	option = {
+		long = "max-bytes"
+		short = "B"
+		type = "integer"
+		help = "keep entries that fit in the given number of bytes"
+                default = "-1"
+	}
+        option = {
+                long = "reset"
+                short = "R"
+                type = "flag"
+                help = "reset the log to version 1; forces full propagation"
+                default = "0";
+        }
 	function = "iprop_truncate"
-	help = "Truncate the log, preserve the version number."
-	max_args = "0"
+	help = "Truncate the log, preserve the version number.  Keeps 100 entries by default."
+	max_args = "1"
 }
 command = {
 	name = "replay"
@@ -101,7 +134,7 @@ command = {
 	}
 	function = "iprop_replay"
 	help = "Replay the log on the database."
-	max_args = "0"
+	max_args = "1"
 }
 command = {
 	name = "last-version"
@@ -112,13 +145,37 @@ command = {
 		help = "configuration file"
 		argument = "file"
 	}
+	option = {
+		long = "no-lock"
+		short = "n"
+		type = "flag"
+		help = "don't lock iprop log"
+	}
+	option = {
+		long = "realm"
+		short = "r"
+		type = "string"
+		help = "realm"
+	}
+	function = "last_version"
+	help = "Print the last version of the log-file."
+}
+command = {
+	name = "signal"
+	option = {
+		long = "config-file"
+		short = "c"
+		type = "string"
+		help = "configuration file"
+		argument = "file"
+	}
 	option = {
 		long = "realm"
 		short = "r"
 		type = "string"
 		help = "realm"
 	}
-	function = "last_version"
+	function = "signal_master"
 	help = "Print the last version of the log-file."
 	max_args = "0"
 }
--- a/lib/kadm5/iprop-log.8
+++ b/lib/kadm5/iprop-log.8
@@ -38,7 +38,7 @@
 .Os Heimdal
 .Sh NAME
 .Nm iprop-log
-.Nd maintain the iprop log file
+.Nd examine and maintain the iprop log file
 .Sh SYNOPSIS
 .Nm
 .Op Fl Fl version
@@ -54,7 +54,20 @@
 .Fl Fl realm= Ns Ar string
 .Xc
 .Oc
+.Oo Fl K Ar integer \*(Ba Xo
+.Fl Fl keep-entries= Ns Ar integer
+.Xc
+.Oc
+.Oo Fl B Ar integer \*(Ba Xo
+.Fl Fl max-bytes= Ns Ar integer
+.Xc
+.Oc
+.Oo Fl R \*(Ba Xo
+.Fl Fl reset
+.Xc
+.Oc
 .Op Fl h | Fl Fl help
+.Op Ar log-file
 .Pp
 .Nm iprop-log dump
 .Oo Fl c Ar file \*(Ba Xo
@@ -65,7 +78,16 @@
 .Fl Fl realm= Ns Ar string
 .Xc
 .Oc
+.Oo Fl n \*(Ba Xo
+.Fl Fl no-lock
+.Xc
+.Oc
+.Oo Fl R \*(Ba Xo
+.Fl Fl reverse
+.Xc
+.Oc
 .Op Fl h | Fl Fl help
+.Op Ar log-file
 .Pp
 .Nm iprop-log replay
 .Op Fl Fl start-version= Ns Ar version-number
@@ -79,6 +101,35 @@
 .Xc
 .Oc
 .Op Fl h | Fl Fl help
+.Op Ar log-file
+.Pp
+.Nm iprop-log last-version
+.Oo Fl c Ar file \*(Ba Xo
+.Fl Fl config-file= Ns Ar file
+.Xc
+.Oc
+.Oo Fl r Ar string \*(Ba Xo
+.Fl Fl realm= Ns Ar string
+.Xc
+.Oc
+.Oo Fl n \*(Ba Xo
+.Fl Fl no-lock
+.Xc
+.Oc
+.Op Fl h | Fl Fl help
+.Op Ar log-files
+.Pp
+.Nm iprop-log signal
+.Oo Fl c Ar file \*(Ba Xo
+.Fl Fl config-file= Ns Ar file
+.Xc
+.Oc
+.Oo Fl r Ar string \*(Ba Xo
+.Fl Fl realm= Ns Ar string
+.Xc
+.Oc
+.Op Fl h | Fl Fl help
+.Pp
 .Sh DESCRIPTION
 Supported options:
 .Bl -tag -width Ds
@@ -94,11 +145,42 @@ command can be one of the following:
 configuration file
 .It Fl r Ar string , Fl Fl realm= Ns Ar string
 realm
+.It Xo
+.Fl K Ar integer ,
+.Fl Fl keep-entries= Ns Ar integer
+.Xc
+.It Xo
+.Fl B Ar integer ,
+.Fl Fl max-bytes= Ns Ar integer
+.Xc
+.It Xo
+.Fl R ,
+.Fl Fl reset
+.Xc
 .El
 .Pp
-Truncates the log. Sets the new logs version number for the to the
-last entry of the old log.  If the log is truncted by emptying the
-file, the log will start over at the first version (0).
+If
+.Fl Fl reset
+is given, then the given, configured, or default log file will be
+truncated and will start at version 1.  This forces full propagations to
+all slave KDCs.
+.Pp
+Otherwise the log will be truncated but some entries will be preserved,
+as specified by the
+.Fl Fl keep-entries
+and/or
+.Fl Fl max-bytes
+options.  The largest number of
+.Fl Fl keep-entries
+entries that are available and fit in the given
+.Fl Fl max-bytes
+option will be used.  The
+.Fl Fl keep-entries option defaults to 100, and the
+.Fl Fl max-bytes
+option defaults to the
+.Ar log-max-size
+parameter in the configuration.
+.Pp
 .It dump
 .Bl -tag -width Ds
 .It Fl c Ar file , Fl Fl config-file= Ns Ar file
@@ -108,9 +190,23 @@ configuration file
 .Fl Fl realm= Ns Ar string
 .Xc
 realm
+.It Xo
+.Fl n Ar string ,
+.Fl Fl no-lock
+.Xc
+.It Xo
+.Fl R Ar string ,
+.Fl Fl reverse
+.Xc
 .El
 .Pp
-Print out all entries in the log to standard output.
+Print out all entries in the given, configured, or default log file to
+standard output.  If the
+.Fl n
+option is used then don't lock the iprop log file.  If the
+.Fl R
+option is used, then print the entries in reverse order
+(this can be useful when the log is very large).
 .It replay
 .Bl -tag -width Ds
 .It Fl Fl start-version= Ns Ar version-number
@@ -125,17 +221,34 @@ configuration file
 realm
 .El
 .Pp
-Replay the changes from specified entries (or all if none is
-specified) in the transaction log to the database.
+Replay the changes from specified entries (or all if none is specified)
+in the given, configured, or default transaction log file to the
+database.
 .It last-version
 .Bl -tag -width Ds
 .It Fl c Ar file , Fl Fl config-file= Ns Ar file
 configuration file
 .It Fl r Ar string , Fl Fl realm= Ns Ar string
 realm
+.It Xo
+.Fl n Ar string ,
+.Fl Fl no-lock
+.Xc
+.El
+.Pp
+prints the version of the last record in each of the given log files, or
+the configured, or the default log file if none is given.
+.It signal
+.Bl -tag -width Ds
+.It Fl c Ar file , Fl Fl config-file= Ns Ar file
+configuration file
+.It Fl r Ar string , Fl Fl realm= Ns Ar string
+realm
 .El
 .Pp
-prints the version of the last log entry.
+Signals the ipropd-master daemon to send updates to slaves.  Normally
+kadmin does this every time it writes to the database, so this should
+rarely be needed.
 .El
 .Sh SEE ALSO
 .Xr iprop 8
--- a/lib/kadm5/iprop-log.c
+++ b/lib/kadm5/iprop-log.c
@@ -46,11 +46,11 @@ get_kadmin_context(const char *config_file, char *realm)
    kadm5_config_params conf;
    krb5_error_code ret;
    void *kadm_handle;
+    char *file = NULL;
    char **files;
    int aret;

    if (config_file == NULL) {
-	char *file;
 	aret = asprintf(&file, "%s/kdc.conf", hdb_db_dir(context));
 	if (aret == -1 || file == NULL)
 	    errx(1, "out of memory");
@@ -58,6 +58,7 @@ get_kadmin_context(const char *config_file, char *realm)
    }

    ret = krb5_prepend_config_files_default(config_file, &files);
+    free(file);
    if (ret)
 	krb5_err(context, 1, ret, "getting configuration files");

@@ -102,7 +103,7 @@ static const char *op_names[] = {
    "nop"
 };

-static void
+static kadm5_ret_t
 print_entry(kadm5_server_context *server_context,
 	    uint32_t ver,
 	    time_t timestamp,
@@ -112,29 +113,28 @@ print_entry(kadm5_server_context *server_context,
 	    void *ctx)
 {
    char t[256];
+    const char *entry_kind = ctx;
    int32_t mask;
+    int32_t nop_time;
+    uint32_t nop_ver;
    hdb_entry ent;
    krb5_principal source;
    char *name1, *name2;
    krb5_data data;
    krb5_context scontext = server_context->context;
+    krb5_error_code ret;

    krb5_data_zero(&data);

-    off_t end = krb5_storage_seek(sp, 0, SEEK_CUR) + len;
-
-    krb5_error_code ret;
-
    strftime(t, sizeof(t), "%Y-%m-%d %H:%M:%S", localtime(&timestamp));

    if((int)op < (int)kadm_get || (int)op > (int)kadm_nop) {
 	printf("unknown op: %d\n", op);
-	krb5_storage_seek(sp, end, SEEK_SET);
-	return;
+	return 0;
    }

-    printf ("%s: ver = %u, timestamp = %s, len = %u\n",
-	    op_names[op], ver, t, len);
+    printf ("%s%s: ver = %u, timestamp = %s, len = %u\n",
+	    entry_kind, op_names[op], ver, t, len);
    switch(op) {
    case kadm_delete:
 	krb5_ret_principal(sp, &source);
@@ -261,12 +261,29 @@ print_entry(kadm5_server_context *server_context,
 	free_hdb_entry(&ent);
 	break;
    case kadm_nop :
+        if (len == 16) {
+            uint64_t off;
+            krb5_ret_uint64(sp, &off);
+            printf("uberblock offset %llu ", (unsigned long long)off);
+        } else {
+            printf("nop");
+        }
+        if (len == 16 || len == 8) {
+            krb5_ret_int32(sp, &nop_time);
+            krb5_ret_uint32(sp, &nop_ver);
+
+            timestamp = nop_time;
+            strftime(t, sizeof(t), "%Y-%m-%d %H:%M:%S", localtime(&timestamp));
+            printf("timestamp %s version %u", t, nop_ver);
+        }
+        printf("\n");
 	break;
    default:
 	abort();
    }
    krb5_data_free(&data);
-    krb5_storage_seek(sp, end, SEEK_SET);
+
+    return 0;
 }

 int
@@ -274,21 +291,61 @@ iprop_dump(struct dump_options *opt, int argc, char **argv)
 {
    kadm5_server_context *server_context;
    krb5_error_code ret;
+    enum kadm_iter_opts iter_opts_1st = 0;
+    enum kadm_iter_opts iter_opts_2nd = 0;
+    char *desc_1st = "";
+    char *desc_2nd = "";

    server_context = get_kadmin_context(opt->config_file_string,
 					opt->realm_string);

-    ret = kadm5_log_init (server_context);
-    if (ret)
-	krb5_err (context, 1, ret, "kadm5_log_init");
+    if (argc > 0) {
+        free(server_context->log_context.log_file);
+        server_context->log_context.log_file = strdup(argv[0]);
+        if (server_context->log_context.log_file == NULL)
+            krb5_err(context, 1, errno, "strdup");
+    }

-    ret = kadm5_log_foreach (server_context, print_entry, NULL);
-    if(ret)
+    if (opt->reverse_flag) {
+        iter_opts_1st = kadm_backward | kadm_unconfirmed;
+        iter_opts_2nd = kadm_backward | kadm_confirmed;
+        desc_1st = "unconfirmed ";
+    } else {
+        iter_opts_1st = kadm_forward | kadm_confirmed;
+        iter_opts_2nd = kadm_forward | kadm_unconfirmed;
+        desc_2nd = "unconfirmed";
+    }
+
+    if (opt->no_lock_flag) {
+        ret = kadm5_log_init_sharedlock(server_context, LOCK_NB);
+        if (ret == EAGAIN || ret == EWOULDBLOCK) {
+            warnx("Not locking the iprop log");
+            ret = kadm5_log_init_nolock(server_context);
+            if (ret)
+                krb5_err(context, 1, ret, "kadm5_log_init_nolock");
+        }
+    } else {
+        warnx("If this command appears to block, try the --no-lock option");
+        ret = kadm5_log_init_sharedlock(server_context, 0);
+        if (ret)
+            krb5_err(context, 1, ret, "kadm5_log_init_sharedlock");
+    }
+
+    ret = kadm5_log_foreach(server_context, iter_opts_1st,
+                            NULL, print_entry, desc_1st);
+    if (ret)
+	krb5_warn(context, ret, "kadm5_log_foreach");
+
+    ret = kadm5_log_foreach(server_context, iter_opts_2nd,
+                            NULL, print_entry, desc_2nd);
+    if (ret)
 	krb5_warn(context, ret, "kadm5_log_foreach");

    ret = kadm5_log_end (server_context);
    if (ret)
 	krb5_warn(context, ret, "kadm5_log_end");
+
+    kadm5_destroy(server_context);
    return 0;
 }

@@ -301,10 +358,38 @@ iprop_truncate(struct truncate_options *opt, int argc, char **argv)
    server_context = get_kadmin_context(opt->config_file_string,
 					opt->realm_string);

-    ret = kadm5_log_truncate (server_context);
-    if (ret)
-	krb5_err (context, 1, ret, "kadm5_log_truncate");
+    if (argc > 0) {
+        free(server_context->log_context.log_file);
+        server_context->log_context.log_file = strdup(argv[0]);
+        if (server_context->log_context.log_file == NULL)
+            krb5_err(context, 1, errno, "strdup");
+    }

+    if (opt->keep_entries_integer < 0 &&
+        opt->max_bytes_integer < 0) {
+        opt->keep_entries_integer = 100;
+        opt->max_bytes_integer = 0;
+    }
+    if (opt->keep_entries_integer < 0)
+        opt->keep_entries_integer = 0;
+    if (opt->max_bytes_integer < 0)
+        opt->max_bytes_integer = 0;
+
+    if (opt->reset_flag) {
+        ret = kadm5_log_reinit(server_context);
+    } else {
+        ret = kadm5_log_init(server_context);
+        if (ret)
+            krb5_err(context, 1, ret, "kadm5_log_init");
+        ret = kadm5_log_truncate(server_context, opt->keep_entries_integer,
+                                 opt->max_bytes_integer);
+    }
+    if (ret)
+	krb5_err(context, 1, ret, "kadm5_log_truncate");
+
+    kadm5_log_signal_master(server_context);
+
+    kadm5_destroy(server_context);
    return 0;
 }

@@ -312,26 +397,72 @@ int
 last_version(struct last_version_options *opt, int argc, char **argv)
 {
    kadm5_server_context *server_context;
+    char *alt_argv[2];
    krb5_error_code ret;
    uint32_t version;
+    size_t i;

    server_context = get_kadmin_context(opt->config_file_string,
 					opt->realm_string);

-    ret = kadm5_log_init (server_context);
-    if (ret)
-	krb5_err (context, 1, ret, "kadm5_log_init");
+    if (argc == 0) {
+        alt_argv[0] = strdup(server_context->log_context.log_file);
+        alt_argv[1] = NULL;
+        if (alt_argv[0] == NULL)
+            krb5_err(context, 1, errno, "strdup");
+        argv = alt_argv;
+        argc = 1;
+    }

-    ret = kadm5_log_get_version (server_context, &version);
-    if (ret)
-	krb5_err (context, 1, ret, "kadm5_log_get_version");
+    for (i = 0; i < argc; i++) {
+        free(server_context->log_context.log_file);
+        server_context->log_context.log_file = strdup(argv[i]);
+        if (server_context->log_context.log_file == NULL)
+            krb5_err(context, 1, errno, "strdup");

-    ret = kadm5_log_end (server_context);
-    if (ret)
-	krb5_warn(context, ret, "kadm5_log_end");
+        if (opt->no_lock_flag) {
+            ret = kadm5_log_init_sharedlock(server_context, LOCK_NB);
+            if (ret == EAGAIN || ret == EWOULDBLOCK) {
+                warnx("Not locking the iprop log");
+                ret = kadm5_log_init_nolock(server_context);
+                if (ret)
+                    krb5_err(context, 1, ret, "kadm5_log_init_nolock");
+            }
+        } else {
+            if (argc == 0)
+                warnx("If this command appears to block, try the "
+                      "--no-lock option");
+            ret = kadm5_log_init_sharedlock(server_context, 0);
+            if (ret)
+                krb5_err(context, 1, ret, "kadm5_log_init_sharedlock");
+        }

-    printf("version: %lu\n", (unsigned long)version);
+        ret = kadm5_log_get_version (server_context, &version);
+        if (ret)
+            krb5_err (context, 1, ret, "kadm5_log_get_version");

+        ret = kadm5_log_end (server_context);
+        if (ret)
+            krb5_warn(context, ret, "kadm5_log_end");
+
+        printf("version: %lu\n", (unsigned long)version);
+    }
+
+    kadm5_destroy(server_context);
+    return 0;
+}
+
+int
+signal_master(struct signal_options *opt, int argc, char **argv)
+{
+    kadm5_server_context *server_context;
+
+    server_context = get_kadmin_context(opt->config_file_string,
+					opt->realm_string);
+
+    kadm5_log_signal_master(server_context);
+
+    kadm5_destroy(server_context);
    return 0;
 }

@@ -342,7 +473,7 @@ last_version(struct last_version_options *opt, int argc, char **argv)
 int start_version = -1;
 int end_version = -1;

-static void
+static kadm5_ret_t
 apply_entry(kadm5_server_context *server_context,
 	    uint32_t ver,
 	    time_t timestamp,
@@ -357,18 +488,18 @@ apply_entry(kadm5_server_context *server_context,
    if((opt->start_version_integer != -1 && ver < (uint32_t)opt->start_version_integer) ||
       (opt->end_version_integer != -1 && ver > (uint32_t)opt->end_version_integer)) {
 	/* XXX skip this entry */
-	krb5_storage_seek(sp, len, SEEK_CUR);
-	return;
+	return 0;
    }
    printf ("ver %u... ", ver);
    fflush (stdout);

-    ret = kadm5_log_replay (server_context,
-			    op, ver, len, sp);
+    ret = kadm5_log_replay(server_context, op, ver, len, sp);
    if (ret)
 	krb5_warn (server_context->context, ret, "kadm5_log_replay");

    printf ("done\n");
+
+    return 0;
 }

 int
@@ -380,6 +511,13 @@ iprop_replay(struct replay_options *opt, int argc, char **argv)
    server_context = get_kadmin_context(opt->config_file_string,
 					opt->realm_string);

+    if (argc > 0) {
+        free(server_context->log_context.log_file);
+        server_context->log_context.log_file = strdup(argv[0]);
+        if (server_context->log_context.log_file == NULL)
+            krb5_err(context, 1, errno, "strdup");
+    }
+
    ret = server_context->db->hdb_open(context,
 				       server_context->db,
 				       O_RDWR | O_CREAT, 0600);
@@ -390,7 +528,9 @@ iprop_replay(struct replay_options *opt, int argc, char **argv)
    if (ret)
 	krb5_err (context, 1, ret, "kadm5_log_init");

-    ret = kadm5_log_foreach (server_context, apply_entry, opt);
+    ret = kadm5_log_foreach(server_context,
+                            kadm_forward | kadm_confirmed | kadm_unconfirmed,
+                            NULL, apply_entry, opt);
    if(ret)
 	krb5_warn(context, ret, "kadm5_log_foreach");
    ret = kadm5_log_end (server_context);
@@ -400,6 +540,7 @@ iprop_replay(struct replay_options *opt, int argc, char **argv)
    if (ret)
 	krb5_err (context, 1, ret, "db->close");

+    kadm5_destroy(server_context);
    return 0;
 }

--- a/lib/kadm5/iprop.8
+++ b/lib/kadm5/iprop.8
@@ -38,7 +38,7 @@
 .Nm iprop ,
 .Nm ipropd-master ,
 .Nm ipropd-slave
-.Nd propagate changes to a Heimdal Kerberos master KDC to slave KDCs
+.Nd propagate transactions from a Heimdal Kerberos master KDC to slave KDCs
 .Sh SYNOPSIS
 .Nm ipropd-master
 .Oo Fl c Ar string \*(Ba Xo
@@ -110,13 +110,18 @@ which sends the whole database to the slaves regularly,
 .Nm
 normally sends only the changes as they happen on the master.
 The master keeps track of all the changes by assigning a version
-number to every change to the database.
+number to every transaction to the database.
 The slaves know which was the latest version they saw, and in this
 way it can be determined if they are in sync or not.
-A log of all the changes is kept on the master.
+A log of all the transactions is kept on the master.
 When a slave is at an older version than the oldest one in the log,
 the whole database has to be sent.
 .Pp
+The log of transactions is also used to implement a two-phase commit
+(with roll-forward for recovery) method of updating the HDB.
+Transactions are first recorded in the log, then in the HDB, then
+the log is updated to mark the transaction as committed.
+.Pp
 The changes are propagated over a secure channel (on port 2121 by
 default).
 This should normally be defined as
@@ -175,6 +180,11 @@ like 5 min, 300 s, or simply a number of seconds.
 .Pa slaves ,
 .Pa slave-stats
 in the database directory.
+.Pa ipropd-master.pid ,
+.Pa ipropd-slave.pid
+in the database directory, or in the directory named by the
+.Ev HEIM_PIDFILE_DIR
+environment variable.
 .Sh SEE ALSO
 .Xr krb5.conf 5 ,
 .Xr hprop 8 ,
--- a/lib/kadm5/ipropd_master.c
+++ b/lib/kadm5/ipropd_master.c
@@ -125,6 +125,7 @@ struct slave {
    char *name;
    krb5_auth_context ac;
    uint32_t version;
+    int32_t version_tstamp;
    time_t seen;
    unsigned long flags;
 #define SLAVE_F_DEAD	0x1
@@ -356,6 +357,9 @@ write_dump (krb5_context context, krb5_storage *dump,
    if (ret)
 	return ret;

+    if (krb5_storage_seek(dump, 0, SEEK_SET) != 0)
+        return errno;
+
    /*
     * First we store zero as the HDB version, this will indicate to a
     * later reader that the dumpfile is invalid.  We later write the
@@ -454,13 +458,15 @@ write_dump (krb5_context context, krb5_storage *dump,

 static int
 send_complete (krb5_context context, slave *s, const char *database,
-	       uint32_t current_version, uint32_t oldest_version)
+	       uint32_t current_version, uint32_t oldest_version,
+	       int32_t initial_log_tstamp)
 {
    krb5_error_code ret;
    krb5_storage *dump = NULL;
    uint32_t vno = 0;
    krb5_data data;
    int fd = -1;
+    struct stat st;
    char *dfn;

    ret = asprintf(&dfn, "%s/ipropd.dumpfile", hdb_db_dir(context));
@@ -506,12 +512,19 @@ send_complete (krb5_context context, slave *s, const char *database,
 	    goto done;
 	}

+        if (fstat(fd, &st) == -1) {
+            ret = errno;
+            krb5_warn(context, ret, "send_complete: could not stat dump file");
+            goto done;
+        }
+
 	/*
 	 * If the current dump has an appropriate version, then we can
 	 * break out of the loop and send the file below.
 	 */

-	if (ret == 0 && vno >= oldest_version && vno <= current_version)
+	if (ret == 0 && vno != 0 && st.st_mtime > initial_log_tstamp &&
+            vno >= oldest_version && vno <= current_version)
 	    break;

 	/*
@@ -546,8 +559,15 @@ send_complete (krb5_context context, slave *s, const char *database,
 	    goto done;
 	}

+        if (fstat(fd, &st) == -1) {
+            ret = errno;
+            krb5_warn(context, ret, "write_dump: could not stat dump file");
+            goto done;
+        }
+
 	/* check if someone wrote a better version for us */
-	if (vno >= oldest_version)
+        if (ret == 0 && vno != 0 && st.st_mtime > initial_log_tstamp &&
+            vno >= oldest_version && vno <= current_version)
 	    continue;

 	/* Now, we know that we must write a new dump file.  */
@@ -643,18 +663,26 @@ send_are_you_there (krb5_context context, slave *s)
 }

 static int
-send_diffs (krb5_context context, slave *s, int log_fd,
-	    const char *database, uint32_t current_version)
+send_diffs (kadm5_server_context *server_context, slave *s, int log_fd,
+	    const char *database, uint32_t current_version,
+	    int32_t current_tstamp)
 {
+    krb5_context context = server_context->context;
    krb5_storage *sp;
-    uint32_t ver;
-    time_t timestamp;
+    uint32_t ver, initial_version, initial_version2;
+    int32_t initial_tstamp, initial_tstamp2;
    enum kadm_ops op;
    uint32_t len;
    off_t right, left;
+    krb5_ssize_t bytes;
    krb5_data data;
    int ret = 0;

+    if (s->flags & SLAVE_F_DEAD) {
+        krb5_warnx(context, "not sending diffs to a dead slave");
+        return 0;
+    }
+
    if (s->version == current_version) {
 	char buf[4];

@@ -666,38 +694,94 @@ send_diffs (krb5_context context, slave *s, int log_fd,
 	data.data   = buf;
 	data.length = 4;
 	ret = krb5_write_priv_message(context, s->ac, &s->fd, &data);
+        if (ret) {
+            krb5_warn(context, ret, "send_diffs: failed to send to slave");
+            slave_dead(context, s);
+        }
 	krb5_warnx(context, "slave %s in sync already at version %ld",
 		   s->name, (long)s->version);
 	return ret;
    }

-    if (s->flags & SLAVE_F_DEAD)
-	return 0;
+    krb5_warnx(context, "sending diffs to a live-seeming slave");

-    flock(log_fd, LOCK_SH);
-    sp = kadm5_log_goto_end (log_fd);
+    /*
+     * XXX The code that makes the diffs should be made a separate function,
+     * then error handling (send_are_you_there() or slave_dead()) can be done
+     * here.
+     */
+
+    if (flock(log_fd, LOCK_SH) == -1) {
+        krb5_warn(context, errno, "could not obtain shared lock on log file");
+        send_are_you_there(context, s);
+        return errno;
+    }
+    ret = kadm5_log_get_version_fd(server_context, log_fd, LOG_VERSION_FIRST,
+                                   &initial_version, &initial_tstamp);
+    sp = kadm5_log_goto_end(server_context, log_fd);
    flock(log_fd, LOCK_UN);
+    if (ret) {
+        if (sp != NULL)
+            krb5_storage_free(sp);
+        krb5_warn(context, ret, "send_diffs: failed to read log");
+        send_are_you_there(context, s);
+        return ret;
+    }
+    if (sp == NULL) {
+        send_are_you_there(context, s);
+        krb5_warn(context, errno ? errno : EINVAL,
+                  "send_diffs: failed to read log");
+        return errno ? errno : EINVAL;
+    }
+    /*
+     * We're not holding any locks here, so we can't prevent truncations.
+     *
+     * We protect against this by re-checking that the initial version and
+     * timestamp are the same before and after this loop.
+     */
    right = krb5_storage_seek(sp, 0, SEEK_CUR);
+    if (right == (off_t)-1) {
+        krb5_storage_free(sp);
+        send_are_you_there(context, s);
+        return errno;
+    }
    for (;;) {
-	ret = kadm5_log_previous (context, sp, &ver, &timestamp, &op, &len);
+	ret = kadm5_log_previous (context, sp, &ver, NULL, &op, &len);
 	if (ret)
 	    krb5_err(context, 1, ret,
 		     "send_diffs: failed to find previous entry");
 	left = krb5_storage_seek(sp, -16, SEEK_CUR);
-	if (ver == s->version)
-	    return 0;
+        if (left == (off_t)-1) {
+            krb5_storage_free(sp);
+            send_are_you_there(context, s);
+            return errno;
+        }
 	if (ver == s->version + 1)
 	    break;
+	if (ver == s->version) {
+            /*
+             * This shouldn't happen, but recall we're not holding a lock on
+             * the log.
+             */
+            krb5_storage_free(sp);
+            krb5_warnx(context, "iprop log truncated while sending diffs to "
+                       "slave??  ver = %lu", (unsigned long)ver);
+            send_are_you_there(context, s);
+            return 0;
+        }
 	if (left == 0) {
 	    krb5_storage_free(sp);
 	    krb5_warnx(context,
 		       "slave %s (version %lu) out of sync with master "
 		       "(first version in log %lu), sending complete database",
 		       s->name, (unsigned long)s->version, (unsigned long)ver);
-	    return send_complete (context, s, database, current_version, ver);
+	    return send_complete (context, s, database, current_version, ver,
+                                  initial_tstamp);
 	}
    }

+    assert(ver == s->version + 1);
+
    krb5_warnx(context,
 	       "syncing slave %s from version %lu to version %lu",
 	       s->name, (unsigned long)s->version,
@@ -707,16 +791,49 @@ send_diffs (krb5_context context, slave *s, int log_fd,
    if (ret) {
 	krb5_storage_free(sp);
 	krb5_warn (context, ret, "send_diffs: krb5_data_alloc");
-	slave_dead(context, s);
+        send_are_you_there(context, s);
 	return 1;
    }
-    krb5_storage_read (sp, (char *)data.data + 4, data.length - 4);
+    bytes = krb5_storage_read(sp, (char *)data.data + 4, data.length - 4);
    krb5_storage_free(sp);
+    if (bytes != data.length - 4) {
+        krb5_warnx(context, "iprop log truncated while sending diffs to "
+                   "slave??  ver = %lu", (unsigned long)ver);
+        send_are_you_there(context, s);
+        return 1;
+    }
+
+    /*
+     * Check that we have the same log initial version and timestamp now as
+     * when we dropped the shared lock on the log file!  Else we could be
+     * sending garbage to the slave.
+     */
+    if (flock(log_fd, LOCK_SH) == -1) {
+        krb5_warn(context, errno, "could not obtain shared lock on log file");
+        send_are_you_there(context, s);
+        return 1;
+    }
+    ret = kadm5_log_get_version_fd(server_context, log_fd, LOG_VERSION_FIRST,
+                                   &initial_version2, &initial_tstamp2);
+    flock(log_fd, LOCK_UN);
+    if (ret) {
+        krb5_warn(context, ret,
+                   "send_diffs: failed to read log while producing diffs");
+        send_are_you_there(context, s);
+        return 1;
+    }
+    if (initial_version != initial_version2 ||
+        initial_tstamp != initial_tstamp2) {
+        krb5_warn(context, ret,
+                   "send_diffs: log truncated while producing diffs");
+        send_are_you_there(context, s);
+        return 1;
+    }

    sp = krb5_storage_from_data (&data);
    if (sp == NULL) {
 	krb5_warnx (context, "send_diffs: krb5_storage_from_data");
-	slave_dead(context, s);
+        send_are_you_there(context, s);
 	return 1;
    }
    krb5_store_int32 (sp, FOR_YOU);
@@ -734,13 +851,17 @@ send_diffs (krb5_context context, slave *s, int log_fd,

    s->version = current_version;

+    krb5_warnx(context, "slave is now up to date");
+
    return 0;
 }

 static int
-process_msg (krb5_context context, slave *s, int log_fd,
-	     const char *database, uint32_t current_version)
+process_msg (kadm5_server_context *server_context, slave *s, int log_fd,
+	     const char *database, uint32_t current_version,
+             int32_t current_tstamp)
 {
+    krb5_context context = server_context->context;
    int ret = 0;
    krb5_data out;
    krb5_storage *sp;
@@ -785,8 +906,9 @@ process_msg (krb5_context context, slave *s, int log_fd,
                       "version we already sent to it");
            s->version = tmp;
 	}
-        ret = send_diffs(context, s, log_fd, database, current_version);
-	break;
+        ret = send_diffs(server_context, s, log_fd, database, current_version,
+                         current_tstamp);
+        break;
    case I_AM_HERE :
 	break;
    case ARE_YOU_THERE:
@@ -968,6 +1090,7 @@ main(int argc, char **argv)
    int log_fd;
    slave *slaves = NULL;
    uint32_t current_version = 0, old_version = 0;
+    int32_t current_tstamp = 0;
    krb5_keytab keytab;
    char **files;
    int aret;
@@ -1053,8 +1176,11 @@ main(int argc, char **argv)
    signal_fd = make_signal_socket (context);
    listen_fd = make_listen_socket (context, port_str);

-    flock(log_fd, LOCK_SH);
-    kadm5_log_get_version_fd (log_fd, &current_version);
+    if (flock(log_fd, LOCK_SH) == -1)
+        krb5_err(context, 1, errno, "shared flock %s",
+                 server_context->log_context.log_file);
+    kadm5_log_get_version_fd(server_context, log_fd, LOG_VERSION_LAST,
+                             &current_version, &current_tstamp);
    flock(log_fd, LOCK_UN);

    krb5_warnx(context, "ipropd-master started at version: %lu",
@@ -1096,10 +1222,42 @@ main(int argc, char **argv)
 		krb5_err (context, 1, errno, "select");
 	}

+        if (stat(server_context->log_context.log_file, &st2) == -1) {
+            krb5_warn(context, errno, "could not stat log file by path");
+            st2 = st;
+        }
+
+        if (st2.st_dev != st.st_dev || st2.st_ino != st.st_ino) {
+            (void) close(log_fd);
+
+            log_fd = open(server_context->log_context.log_file, O_RDONLY, 0);
+            if (log_fd < 0)
+                krb5_err(context, 1, 1, "open %s",
+                          server_context->log_context.log_file);
+
+            if (fstat(log_fd, &st) == -1)
+                krb5_err(context, 1, errno, "stat %s",
+                         server_context->log_context.log_file);
+
+            if (flock(log_fd, LOCK_SH) == -1)
+                krb5_err(context, 1, errno, "shared flock %s",
+                         server_context->log_context.log_file);
+            kadm5_log_get_version_fd(server_context, log_fd, LOG_VERSION_LAST,
+                                     &current_version, &current_tstamp);
+            flock(log_fd, LOCK_UN);
+        }
+
 	if (ret == 0) {
-	    old_version = current_version;
-	    flock(log_fd, LOCK_SH);
-	    kadm5_log_get_version_fd (log_fd, &current_version);
+            /* Recover from failed transactions */
+            if (kadm5_log_init_nb(server_context) == 0)
+                kadm5_log_end(server_context);
+
+	    if (flock(log_fd, LOCK_SH) == -1) {
+                krb5_err(context, 1, errno,
+                         "could not lock log file");
+            }
+	    kadm5_log_get_version_fd(server_context, log_fd, LOG_VERSION_LAST,
+                                     &current_version, &current_tstamp);
 	    flock(log_fd, LOCK_UN);

 	    if (current_version > old_version) {
@@ -1110,7 +1268,8 @@ main(int argc, char **argv)
 		for (p = slaves; p != NULL; p = p->next) {
 		    if (p->flags & SLAVE_F_DEAD)
 			continue;
-		    send_diffs (context, p, log_fd, database, current_version);
+		    send_diffs (server_context, p, log_fd, database,
+                                current_version, current_tstamp);
 		}
 	    }
 	}
@@ -1131,10 +1290,25 @@ main(int argc, char **argv)
 	    --ret;
 	    assert(ret >= 0);
 	    old_version = current_version;
-	    flock(log_fd, LOCK_SH);
-	    kadm5_log_get_version_fd (log_fd, &current_version);
+	    if (flock(log_fd, LOCK_SH) == -1)
+                krb5_err(context, 1, errno, "shared flock %s",
+                         server_context->log_context.log_file);
+	    kadm5_log_get_version_fd(server_context, log_fd, LOG_VERSION_LAST,
+                                     &current_version, &current_tstamp);
 	    flock(log_fd, LOCK_UN);
-	    if (current_version > old_version) {
+	    if (current_version != old_version) {
+                /*
+                 * If current_version < old_version then the log got
+                 * truncated and we'll end up doing full propagations.
+                 *
+                 * Truncating the log when the current version is
+                 * numerically small can lead to race conditions.
+                 * Ideally we should identify log versions as
+                 * {init_or_trunc_time, vno}, then we could not have any
+                 * such race conditions, but this would either require
+                 * breaking backwards compatibility for the protocol or
+                 * adding new messages to it.
+                 */
 		krb5_warnx(context,
 			   "Got a signal, updating slaves %lu to %lu",
 			   (unsigned long)old_version,
@@ -1142,7 +1316,8 @@ main(int argc, char **argv)
 		for (p = slaves; p != NULL; p = p->next) {
 		    if (p->flags & SLAVE_F_DEAD)
 			continue;
-		    send_diffs (context, p, log_fd, database, current_version);
+		    send_diffs (server_context, p, log_fd, database,
+                                current_version, current_tstamp);
 		}
 	    } else {
 		krb5_warnx(context,
@@ -1157,7 +1332,8 @@ main(int argc, char **argv)
 	    if (ret && FD_ISSET(p->fd, &readset)) {
 		--ret;
 		assert(ret >= 0);
-		if(process_msg (context, p, log_fd, database, current_version))
+		if(process_msg (server_context, p, log_fd, database,
+				current_version, current_tstamp))
 		    slave_dead(context, p);
 	    } else if (slave_gone_p (p))
 		slave_dead(context, p);
--- a/lib/kadm5/ipropd_slave.c
+++ b/lib/kadm5/ipropd_slave.c
@@ -176,97 +176,184 @@ ihave(krb5_context context, krb5_auth_context auth_context,
    return ret;
 }

-static void
+static int
 receive_loop (krb5_context context,
 	      krb5_storage *sp,
 	      kadm5_server_context *server_context)
 {
    int ret;
-    off_t left, right;
+    off_t left, right, off;
+    size_t mlen;
    void *buf;
-    int32_t vers, vers2;
-    ssize_t sret;
+    int32_t len, vers, vers2;
+    ssize_t sret, smlen;

    /*
-     * Seek to the current version of the local database.
+     * Seek to the first entry in the message from the master that is
+     * past the current version of the local database.
     */
    do {
-	int32_t len, timestamp, tmp;
+	int32_t timestamp, tmp;

-	if(krb5_ret_int32 (sp, &vers) != 0)
-	    return;
-	krb5_ret_int32 (sp, &timestamp);
-	krb5_ret_int32 (sp, &tmp);
-	krb5_ret_int32 (sp, &len);
-	if ((uint32_t)vers <= server_context->log_context.version)
-	    krb5_storage_seek(sp, len + 8, SEEK_CUR);
+	if (krb5_ret_int32(sp, &vers) != 0 ||
+            krb5_ret_int32(sp, &timestamp) != 0 ||
+            krb5_ret_int32(sp, &tmp) != 0 ||
+            krb5_ret_int32(sp, &len) != 0) {
+
+            /*
+             * This shouldn't happen.  Reconnecting probably won't help
+             * if it does happen, but by reconnecting we get a chance to
+             * connect to a new master if a new one is configured.
+             */
+            krb5_warnx(context, "iprop entries from master were truncated");
+            return EINVAL;
+        }
+        if (len < 0) {
+            krb5_warnx(context, "master sent entry with negative length for"
+                       "version %ld", (long)vers);
+            return EINVAL;
+        }
+	if ((uint32_t)vers > server_context->log_context.version)
+            break;
+        off = krb5_storage_seek(sp, 0, SEEK_CUR);
+        if (krb5_storage_seek(sp, len + 8, SEEK_CUR) != off + len + 8) {
+            krb5_warnx(context, "iprop entries from master were truncated");
+            return 0;
+        }
    } while((uint32_t)vers <= server_context->log_context.version);

    /*
-     * Read up rest of the entires into the memory...
+     * Read the remaining entries into memory...
     */
-    left  = krb5_storage_seek (sp, -16, SEEK_CUR);
-    right = krb5_storage_seek (sp, 0, SEEK_END);
-    buf = malloc (right - left);
-    if (buf == NULL && (right - left) != 0)
-	krb5_errx (context, 1, "malloc: no memory");
+    /* SEEK_CUR is a header into the first entry we care about */
+    left  = krb5_storage_seek(sp, -16, SEEK_CUR);
+    right = krb5_storage_seek(sp, 0, SEEK_END);
+    if (right - left < 24 + len) {
+        krb5_warnx(context, "iprop entries from master were truncated");
+        return EINVAL;
+    }
+
+    mlen = (size_t)(right - left);
+    smlen = right - left;
+    buf = malloc (mlen);
+    if (buf == NULL && mlen != 0) {
+	krb5_warn(context, errno, "malloc: no memory");
+        return ENOMEM;
+    }

    /*
     * ...and then write them out to the on-disk log.
     */
-    krb5_storage_seek (sp, left, SEEK_SET);
-    krb5_storage_read (sp, buf, right - left);
-    sret = write (server_context->log_context.log_fd, buf, right-left);
-    if (sret != right - left)
+    /* NOTE: We haven't validated the entries yet */
+    if (krb5_storage_seek(sp, left, SEEK_SET) != left)
+        krb5_errx(context, 1, "krb5_storage_seek() failed");
+    sret = krb5_storage_read(sp, buf, mlen);
+    if (sret < 0)
+        return errno;
+    if (mlen != (size_t)sret)
+        krb5_errx(context, 1, "short krb5_storage_read() from memory buffer");
+    sret = write(server_context->log_context.log_fd, buf, mlen);
+    if (sret != smlen) {
+        /* This is probably ENOSPC.  We can't recover. */
 	krb5_err(context, 1, errno, "Failed to write log to disk");
-    ret = fsync (server_context->log_context.log_fd);
-    if (ret)
+    }
+    ret = fsync(server_context->log_context.log_fd);
+    if (ret) {
+        /* This is also probably ENOSPC.  We can't recover. */
 	krb5_err(context, 1, errno, "Failed to sync log to disk");
-    free (buf);
+    }
+    free(buf);

    /*
-     * Go back to the startpoint and start to commit the entires to
-     * the database.
+     * Go back to the startpoint and commit the entries to the HDB.
     */
-    krb5_storage_seek (sp, left, SEEK_SET);
+    krb5_storage_seek(sp, left, SEEK_SET);
+    ret = kadm5_log_recover(server_context, kadm_recover_replay);
+    if (ret) {
+        krb5_warn(context, ret, "replay of entries from master failed");
+        return ret;
+    }

-    for(;;) {
-	int32_t len, len2, timestamp, tmp;
+    for (;;) {
+	int32_t len2, timestamp, tmp;
 	off_t cur, cur2;
 	enum kadm_ops op;

-	if(krb5_ret_int32 (sp, &vers) != 0)
-	    break;
-	ret = krb5_ret_int32 (sp, &timestamp);
-	if (ret) krb5_errx(context, 1, "entry %ld: too short", (long)vers);
-	ret = krb5_ret_int32 (sp, &tmp);
-	if (ret) krb5_errx(context, 1, "entry %ld: too short", (long)vers);
+	if (krb5_ret_int32(sp, &vers) != 0)
+            break;
+	ret = krb5_ret_int32(sp, &timestamp);
+	if (ret) {
+            krb5_warnx(context, "entry %ld: too short", (long)vers);
+            return EINVAL;
+        }
+	ret = krb5_ret_int32(sp, &tmp);
+	if (ret) {
+            krb5_warnx(context, "entry %ld: too short", (long)vers);
+            return EINVAL;
+        }
 	op = tmp;
-	ret = krb5_ret_int32 (sp, &len);
-	if (ret) krb5_errx(context, 1, "entry %ld: too short", (long)vers);
-	if (len < 0)
-	    krb5_errx(context, 1, "log is corrupted, "
-		      "negative length of entry version %ld: %ld",
-		      (long)vers, (long)len);
+	ret = krb5_ret_int32(sp, &len);
+	if (ret) {
+            krb5_warnx(context, "entry %ld: too short", (long)vers);
+            return EINVAL;
+        }
+	if (len < 0) {
+	    krb5_warnx(context, "entry %ld: negative length (%ld); "
+                       "master is confused", (long)vers, (long)len);
+            return EINVAL;
+        }
 	cur = krb5_storage_seek(sp, 0, SEEK_CUR);

-	krb5_warnx (context, "replaying entry %d", (int)vers);
+	krb5_warnx(context, "replaying entry %d", (int)vers);

-	ret = kadm5_log_replay (server_context,
-				op, vers, len, sp);
+        /*
+         * kadm5_log_replay() returns errors from among others, the HDB
+         * layer, which can return errors from the actual DBs, some of
+         * which return -1 and set errno, and some of which return
+         * system error codes.
+         */
+	ret = kadm5_log_replay(server_context,
+                               op, vers, len, sp);
+        if (ret == -1 && errno != 0)
+            ret = errno;
 	if (ret) {
-	    const char *s = krb5_get_error_message(server_context->context, ret);
-	    krb5_warnx (context,
-			"kadm5_log_replay: %ld. Lost entry entry, "
-			"Database out of sync ?: %s (%d)",
-			(long)vers, s ? s : "unknown error", ret);
+            const char *s = krb5_get_error_message(server_context->context, ret);
+
+            /*
+             * XXX We don't really know here whether the error is
+             * recoverable or not.  Some HDB errors might be safe to
+             * ignore, and others will not be (e.g., any resulting from
+             * ENOSPC), but we can't tell which is which, particularly
+             * as errors from the databases are not mapped to HDB_ERR_*.
+             *
+             * We do our best to die if the error is not recoverable.
+             */
+            switch (ret) {
+#ifdef EDQUOT
+            case EDQUOT:
+#endif
+            case ENOSPC:
+            case EPIPE:
+            case EINTR:
+            case EFBIG:
+            case EIO:
+                krb5_err(context, 1, ret, "kadm5_log_replay: %ld. Fatal write "
+                         "error: %s (%d)", (long)vers,
+                         s ? s : "unknown error", ret);
+            }
+
+            krb5_warnx(context,
+                       "kadm5_log_replay: %ld. Replay failed. "
+                       "Database out of sync?: %s (%d)",
+                       (long)vers, s ? s : "unknown error", ret);
 	    krb5_free_error_message(context, s);
 	}

 	{
 	    /*
-	     * Make sure the krb5_log_replay does the right thing wrt
-	     * reading out data from the sp.
+	     * Make sure that kadm5_log_replay() read the whole entry
+             * from sp and left the sp offset at the start of the
+             * trailer.
 	     */
 	    cur2 = krb5_storage_seek(sp, 0, SEEK_CUR);
 	    if (cur + len != cur2)
@@ -275,59 +362,76 @@ receive_loop (krb5_context context,
 			  (long)vers);
 	}

-	if (krb5_ret_int32 (sp, &len2) != 0)
-	    krb5_errx(context, 1, "entry %ld: postamble too short", (long)vers);
-	if(krb5_ret_int32 (sp, &vers2) != 0)
-	    krb5_errx(context, 1, "entry %ld: postamble too short", (long)vers);
+	if (krb5_ret_int32(sp, &len2) != 0) {
+	    krb5_warnx(context, "entry %ld: postamble too short; "
+                       "master is confused", (long)vers);
+            return EINVAL;
+        }
+	if(krb5_ret_int32(sp, &vers2) != 0) {
+	    krb5_warnx(context, "entry %ld: postamble too short; "
+                       "master is confused", (long)vers);
+            return EINVAL;
+        }
+	if (len != len2) {
+	    krb5_warnx(context, "entry %ld: len != len2; master is "
+                       "confused", (long)vers);
+            return EINVAL;
+        }
+	if (vers != vers2) {
+	    krb5_warnx(context, "entry %ld: vers != vers2; master is "
+                       "confused", (long)vers);
+            return EINVAL;
+        }

-	if (len != len2)
-	    krb5_errx(context, 1, "entry %ld: len != len2", (long)vers);
-	if (vers != vers2)
-	    krb5_errx(context, 1, "entry %ld: vers != vers2", (long)vers);
+        /*
+         * Update version after each replay.
+         */
+        server_context->log_context.version = vers;
+        kadm5_log_update_uber(server_context);
    }

-    /*
-     * Update version
-     */
-
-    server_context->log_context.version = vers;
+    return 0;
 }

-static void
-receive (krb5_context context,
-	 krb5_storage *sp,
-	 kadm5_server_context *server_context)
+static int
+receive(krb5_context context,
+        krb5_storage *sp,
+        kadm5_server_context *server_context)
 {
-    int ret;
+    krb5_error_code ret, ret2;

    ret = server_context->db->hdb_open(context,
 				       server_context->db,
 				       O_RDWR | O_CREAT, 0600);
    if (ret)
-	krb5_err (context, 1, ret, "db->open");
+	krb5_err(context, 1, ret, "db->open");

-    receive_loop (context, sp, server_context);
+    ret2 = receive_loop(context, sp, server_context);
+    if (ret2)
+	krb5_warn(context, ret, "receive from ipropd-master had errors");

-    ret = server_context->db->hdb_close (context, server_context->db);
+    ret = server_context->db->hdb_close(context, server_context->db);
    if (ret)
-	krb5_err (context, 1, ret, "db->close");
+	krb5_err(context, 1, ret, "db->close");
+
+    return ret2;
 }

 static void
-send_im_here (krb5_context context, int fd,
-	      krb5_auth_context auth_context)
+send_im_here(krb5_context context, int fd,
+	     krb5_auth_context auth_context)
 {
    krb5_storage *sp;
    krb5_data data;
-    int ret;
+    krb5_error_code ret;

-    ret = krb5_data_alloc (&data, 4);
+    ret = krb5_data_alloc(&data, 4);
    if (ret)
-	krb5_err (context, 1, ret, "send_im_here");
+	krb5_err(context, 1, ret, "send_im_here");

    sp = krb5_storage_from_data (&data);
    if (sp == NULL)
-	krb5_errx (context, 1, "krb5_storage_from_data");
+	krb5_errx(context, 1, "krb5_storage_from_data");
    krb5_store_int32(sp, I_AM_HERE);
    krb5_storage_free(sp);

@@ -335,7 +439,9 @@ send_im_here (krb5_context context, int fd,
    krb5_data_free(&data);

    if (ret)
-	krb5_err (context, 1, ret, "krb5_write_priv_message");
+	krb5_err(context, 1, ret, "krb5_write_priv_message");
+
+    return;
 }

 static void
@@ -347,15 +453,7 @@ reinit_log(krb5_context context,

    ret = kadm5_log_reinit(server_context);
    if (ret)
-	krb5_err(context, 1, ret, "kadm5_log_reinit");
-
-    ret = kadm5_log_set_version(server_context, vno - 1);
-    if (ret)
-	krb5_err(context, 1, ret, "kadm5_log_set_version");
-
-    ret = kadm5_log_nop(server_context);
-    if (ret)
-	krb5_err(context, 1, ret, "kadm5_log_nop");
+        krb5_err(context, 1, ret, "kadm5_log_reinit");
 }


@@ -557,7 +655,7 @@ usage(int status)
 int
 main(int argc, char **argv)
 {
-    krb5_error_code ret;
+    krb5_error_code ret, ret2;
    krb5_context context;
    krb5_auth_context auth_context;
    void *kadm_handle;
@@ -763,31 +861,56 @@ main(int argc, char **argv)
 		    krb5_err (context, 1, errno, "select");
 	    }
 	    if (ret == 0) {
-		krb5_warn (context, 1, "server didn't send a message "
-			   "in %d seconds", time_before_lost);
+		krb5_warnx(context, "server didn't send a message "
+                           "in %d seconds", time_before_lost);
 		connected = FALSE;
 		continue;
 	    }

 	    ret = krb5_read_priv_message(context, auth_context, &master_fd, &out);
 	    if (ret) {
-		krb5_warn (context, ret, "krb5_read_priv_message");
+		krb5_warn(context, ret, "krb5_read_priv_message");
 		connected = FALSE;
 		continue;
 	    }

 	    sp = krb5_storage_from_mem (out.data, out.length);
-	    krb5_ret_int32(sp, &tmp);
+            if (sp == NULL)
+                krb5_err(context, 1, errno, "krb5_storage_from_mem");
+	    ret = krb5_ret_int32(sp, &tmp);
+            if (ret == HEIM_ERR_EOF) {
+                krb5_warn(context, ret, "master sent zero-length message");
+                connected = FALSE;
+                continue;
+            }
+            if (ret != 0) {
+                krb5_warn(context, ret, "couldn't read master's message");
+                connected = FALSE;
+                continue;
+            }
+
+            ret = kadm5_log_init(server_context);
+            if (ret) {
+                krb5_err(context, 1, ret, "kadm5_log_init while handling a "
+                         "message from the master");
+            }
 	    switch (tmp) {
 	    case FOR_YOU :
-		receive(context, sp, server_context);
+		ret2 = receive(context, sp, server_context);
+                if (ret2)
+                    krb5_warn(context, ret,
+                              "receive from ipropd-master had errors");
 		ret = ihave(context, auth_context, master_fd,
 			    server_context->log_context.version);
-		if (ret)
+		if (ret || ret2)
 		    connected = FALSE;
-		else
-		    is_up_to_date(context, status_file, server_context);

+                /*
+                 * If it returns an error, receive() may nonetheless
+                 * have committed some entries successfully, so we must
+                 * update the slave_status even if there were errors.
+                 */
+                is_up_to_date(context, status_file, server_context);
 		break;
 	    case TELL_YOU_EVERYTHING :
 		ret = receive_everything(context, master_fd, server_context,
@@ -803,6 +926,11 @@ main(int argc, char **argv)
 		break;
 	    case ARE_YOU_THERE :
 		is_up_to_date(context, status_file, server_context);
+                ret = ihave(context, auth_context, master_fd,
+                            server_context->log_context.version);
+                if (ret)
+                    connected = FALSE;
+
 		send_im_here(context, master_fd, auth_context);
 		break;
 	    case YOU_HAVE_LAST_VERSION:
@@ -839,8 +967,10 @@ main(int argc, char **argv)
 	}
    }

-    if (status_file)
+    if (status_file) {
+        /* XXX It'd be better to leave it saying we're not here */
 	unlink(status_file);
+    }

    if (0);
 #ifndef NO_SIGXCPU
--- a/lib/kadm5/kadm5_err.et
+++ b/lib/kadm5/kadm5_err.et
@@ -65,3 +65,4 @@ error_code KEEPOLD_NOSUPP,	"Keep old keys option not supported"
 error_code AUTH_GET_KEYS,	"Operation requires `get-keys' privilege"
 error_code ALREADY_LOCKED,	"Database already locked"
 error_code NOT_LOCKED,		"Database not locked"
+error_code LOG_CORRUPT,		"Incremental propagation log got corrupted"
--- a/lib/kadm5/libkadm5srv-exports.def
+++ b/lib/kadm5/libkadm5srv-exports.def
@@ -59,6 +59,7 @@ EXPORTS
 	kadm5_s_init_with_creds
 	kadm5_s_chpass_principal_cond
 	kadm5_log_set_version
+	kadm5_log_signal_master
 ;!	kadm5_log_signal_socket
 	kadm5_log_signal_socket_info    ;!
 	kadm5_log_previous
@@ -66,10 +67,14 @@ EXPORTS
 	kadm5_log_foreach
 	kadm5_log_get_version_fd
 	kadm5_log_get_version
+	kadm5_log_recover
 	kadm5_log_replay
 	kadm5_log_end
 	kadm5_log_reinit
 	kadm5_log_init
+	kadm5_log_init_nb
+	kadm5_log_init_nolock
+	kadm5_log_init_sharedlock
 	kadm5_log_nop
 	kadm5_log_truncate
 	kadm5_log_modify
--- a/lib/kadm5/log.c
+++ b/lib/kadm5/log.c
--- a/lib/kadm5/modify_s.c
+++ b/lib/kadm5/modify_s.c
@@ -57,6 +57,11 @@ modify_principal(void *server_handle,
 	if(ret)
 	    return ret;
    }
+
+    ret = kadm5_log_init(context);
+    if (ret)
+        goto out;
+
    ret = context->db->hdb_fetch_kvno(context->context, context->db,
 				      princ->principal, HDB_F_GET_ANY|HDB_F_ADMIN_DATA, 0, &ent);
    if(ret)
@@ -105,20 +110,20 @@ modify_principal(void *server_handle,
 	    goto out2;
    }

-    ret = context->db->hdb_store(context->context, context->db,
-			     HDB_F_REPLACE, &ent);
-    if (ret)
-	goto out2;
-
-    kadm5_log_modify (context,
-		      &ent.entry,
-		      mask | KADM5_MOD_NAME | KADM5_MOD_TIME);
+    /* This logs the change for iprop and writes to the HDB */
+    ret = kadm5_log_modify(context, &ent.entry,
+                           mask | KADM5_MOD_NAME | KADM5_MOD_TIME);

 out2:
    hdb_free_entry(context->context, &ent);
 out:
-    if (!context->keep_open)
-	context->db->hdb_close(context->context, context->db);
+    (void) kadm5_log_end(context);
+    if (!context->keep_open) {
+        kadm5_ret_t ret2;
+        ret2 = context->db->hdb_close(context->context, context->db);
+        if (ret == 0 && ret2 != 0)
+            ret = ret2;
+    }
    return _kadm5_error_code(ret);
 }

--- a/lib/kadm5/private.h
+++ b/lib/kadm5/private.h
@@ -82,7 +82,10 @@ typedef struct kadm5_log_peer {
 typedef struct kadm5_log_context {
    char *log_file;
    int log_fd;
+    int read_only;
+    int lock_mode;
    uint32_t version;
+    time_t last_time;
 #ifndef NO_UNIX_SOCKETS
    struct sockaddr_un socket_name;
 #else
@@ -136,6 +139,11 @@ typedef struct kadm5_ad_context {
    char *base_dn;
 } kadm5_ad_context;

+/*
+ * This enum is used in the iprop log file and on the wire in the iprop
+ * protocol.  DO NOT CHANGE, except to add new op types at the end, and
+ * look for places in lib/kadm5/log.c to update.
+ */
 enum kadm_ops {
    kadm_get,
    kadm_delete,
@@ -147,7 +155,28 @@ enum kadm_ops {
    kadm_get_privs,
    kadm_get_princs,
    kadm_chpass_with_key,
-    kadm_nop
+    kadm_nop,
+    kadm_first = kadm_get,
+    kadm_last = kadm_nop
+};
+
+/* FIXME nop types are currently not implemented */
+enum kadm_nop_type {
+    kadm_nop_plain, /* plain nop, not relevance except as uberblock */
+    kadm_nop_trunc, /* indicates that the master truncated the log  */
+    kadm_nop_close  /* indicates that the master closed this log    */
+};
+
+enum kadm_iter_opts {
+    kadm_forward        = 1,
+    kadm_backward       = 2,
+    kadm_confirmed      = 4,
+    kadm_unconfirmed    = 8
+};
+
+enum kadm_recover_mode {
+    kadm_recover_commit,
+    kadm_recover_replay
 };

 #define KADMIN_APPL_VERSION "KADM0.1"
--- a/lib/kadm5/randkey_s.c
+++ b/lib/kadm5/randkey_s.c
@@ -59,6 +59,11 @@ kadm5_s_randkey_principal(void *server_handle,
 	if(ret)
 	    return ret;
    }
+
+    ret = kadm5_log_init(context);
+    if (ret)
+        goto out;
+
    ret = context->db->hdb_fetch_kvno(context->context, context->db, princ,
 				      HDB_F_GET_ANY|HDB_F_ADMIN_DATA, 0, &ent);
    if(ret)
@@ -70,12 +75,8 @@ kadm5_s_randkey_principal(void *server_handle,
 	    goto out2;
    }

-    ret = _kadm5_set_keys_randomly (context,
-				    &ent.entry,
-				    n_ks_tuple,
-				    ks_tuple,
-				    new_keys,
-				    n_keys);
+    ret = _kadm5_set_keys_randomly(context, &ent.entry, n_ks_tuple, ks_tuple,
+                                   new_keys, n_keys);
    if (ret)
 	goto out2;
    ent.entry.kvno++;
@@ -104,22 +105,19 @@ kadm5_s_randkey_principal(void *server_handle,
 	hdb_replace_extension(context->context, &ent.entry, &ext);
    }

-    ret = context->db->hdb_store(context->context, context->db,
-				 HDB_F_REPLACE, &ent);
-    if (ret)
-	goto out2;
-
-    kadm5_log_modify(context, &ent.entry,
-		     KADM5_ATTRIBUTES | KADM5_PRINCIPAL | KADM5_MOD_NAME |
-		     KADM5_MOD_TIME | KADM5_KEY_DATA | KADM5_KVNO |
-		     KADM5_PW_EXPIRATION | KADM5_TL_DATA);
+    /* This logs the change for iprop and writes to the HDB */
+    ret = kadm5_log_modify(context, &ent.entry,
+                           KADM5_ATTRIBUTES | KADM5_PRINCIPAL |
+                           KADM5_MOD_NAME | KADM5_MOD_TIME |
+                           KADM5_KEY_DATA | KADM5_KVNO |
+                           KADM5_PW_EXPIRATION | KADM5_TL_DATA);

 out3:
    if (ret) {
 	int i;

 	for (i = 0; i < *n_keys; ++i)
-	    krb5_free_keyblock_contents (context->context, &(*new_keys)[i]);
+	    krb5_free_keyblock_contents(context->context, &(*new_keys)[i]);
 	free (*new_keys);
 	*new_keys = NULL;
 	*n_keys = 0;
@@ -127,7 +125,12 @@ out3:
 out2:
    hdb_free_entry(context->context, &ent);
 out:
-    if (!context->keep_open)
-	context->db->hdb_close(context->context, context->db);
+    (void) kadm5_log_end(context);
+    if (!context->keep_open) {
+        kadm5_ret_t ret2;
+        ret2 = context->db->hdb_close(context->context, context->db);
+        if (ret == 0 && ret2 != 0)
+            ret = ret2;
+    }
    return _kadm5_error_code(ret);
 }
--- a/lib/kadm5/rename_s.c
+++ b/lib/kadm5/rename_s.c
@@ -53,15 +53,18 @@ kadm5_s_rename_principal(void *server_handle,
 	if(ret)
 	    return ret;
    }
+
+    ret = kadm5_log_init(context);
+    if (ret)
+        goto out;
+
    ret = context->db->hdb_fetch_kvno(context->context, context->db,
 				      source, HDB_F_GET_ANY|HDB_F_ADMIN_DATA, 0, &ent);
-    if(ret){
-	if (!context->keep_open)
-	    context->db->hdb_close(context->context, context->db);
+    if (ret)
 	goto out;
-    }
+    oldname = ent.entry.principal;
    ret = _kadm5_set_modifier(context, &ent.entry);
-    if(ret)
+    if (ret)
 	goto out2;
    {
 	/* fix salt */
@@ -85,31 +88,30 @@ kadm5_s_rename_principal(void *server_handle,
 	}
 	krb5_free_salt(context->context, salt2);
    }
-    if(ret)
+    if (ret)
 	goto out2;
-    oldname = ent.entry.principal;
+
+    /* Borrow target */
    ent.entry.principal = target;
-
    ret = hdb_seal_keys(context->context, context->db, &ent.entry);
-    if (ret) {
-	ent.entry.principal = oldname;
+    if (ret)
 	goto out2;
-    }

-    kadm5_log_rename (context, source, &ent.entry);
+    /* This logs the change for iprop and writes to the HDB */
+    ret = kadm5_log_rename(context, source, &ent.entry);

-    ret = context->db->hdb_store(context->context, context->db, 0, &ent);
-    if(ret){
-	ent.entry.principal = oldname;
-	goto out2;
-    }
-    ret = context->db->hdb_remove(context->context, context->db, oldname);
-    ent.entry.principal = oldname;
 out2:
-    if (!context->keep_open)
-	context->db->hdb_close(context->context, context->db);
+    ent.entry.principal = oldname; /* Unborrow target */
    hdb_free_entry(context->context, &ent);
+
 out:
+    (void) kadm5_log_end(context);
+    if (!context->keep_open) {
+        kadm5_ret_t ret2;
+        ret2 = context->db->hdb_close(context->context, context->db);
+        if (ret == 0 && ret2 != 0)
+            ret = ret2;
+    }
    return _kadm5_error_code(ret);
 }

--- a/lib/kadm5/version-script.map
+++ b/lib/kadm5/version-script.map
@@ -62,16 +62,21 @@ HEIMDAL_KAMD5_SERVER_1.0 {
 		kadm5_s_init_with_creds;
 		kadm5_s_chpass_principal_cond;
 		kadm5_log_set_version;
+		kadm5_log_signal_master;
 		kadm5_log_signal_socket;
 		kadm5_log_previous;
 		kadm5_log_goto_end;
 		kadm5_log_foreach;
 		kadm5_log_get_version_fd;
 		kadm5_log_get_version;
+		kadm5_log_recover;
 		kadm5_log_replay;
 		kadm5_log_end;
 		kadm5_log_reinit;
 		kadm5_log_init;
+		kadm5_log_init_nb;
+		kadm5_log_init_nolock;
+		kadm5_log_init_sharedlock;
 		kadm5_log_nop;
 		kadm5_log_truncate;
 		kadm5_log_modify;
--- a/lib/krb5/krb5.conf.5
+++ b/lib/krb5/krb5.conf.5
@@ -573,7 +573,25 @@ Use this file for the ACL list of this database.
 Use this file as the log of changes performed to the database.
 This file is used by
 .Nm ipropd-master
-for propagating changes to slaves.
+for propagating changes to slaves.  It is also used by
+.Nm kadmind
+and
+.Nm kadmin
+(when used with the
+.Li -l
+option), and by all applications using
+.Nm libkadm5
+with the local backend, for two-phase commit functionality.  Slaves also
+use this.  Setting this to
+.Nm /dev/null
+disables two-phase commit and incremental propagation.  Use
+.Nm iprop-log
+to show the contents of this log file.
+.It Li log-max-size = Pa number
+When the log reaches this size (in bytes), the log will be truncated,
+saving some entries, and keeping the latest version number so as to not
+disrupt incremental propagation.  If set to a negative value then
+automatic log truncation will be disabled.  Defaults to 52428800 (50MB).
 .El
 .It Li }
 .It Li max-request = Va SIZE
--- a/lib/krb5/libkrb5-exports.def.in
+++ b/lib/krb5/libkrb5-exports.def.in
@@ -751,6 +751,7 @@ EXPORTS
 	_krb5_dh_group_ok
 	_krb5_get_host_realm_int
 	_krb5_get_int
+        _krb5_get_int64
 	_krb5_pac_sign
 	_krb5_parse_moduli
 	_krb5_pk_kdf
--- a/lib/krb5/version-script.map
+++ b/lib/krb5/version-script.map
@@ -744,6 +744,7 @@ HEIMDAL_KRB5_2.0 {
 		_krb5_dh_group_ok;
 		_krb5_get_host_realm_int;
 		_krb5_get_int;
+		_krb5_get_int64;
 		_krb5_pac_sign;
 		_krb5_parse_moduli;
 		_krb5_pk_kdf;
--- a/tests/kdc/check-iprop.in
+++ b/tests/kdc/check-iprop.in
@@ -86,6 +86,35 @@ ${kadmin} -l ext -k ${keytab} iprop/slave.test.h5l.se@${R} || exit 1

 echo foo > ${objdir}/foopassword

+echo "Test log recovery"
+${kadmin} -l add --random-key --use-defaults recovtest@${R} || exit 1
+# Test theory: save the log, make a change and save the record it
+# produced, restore the log, append to it the saved record, then get
+
+# Save the log
+cp current.log current.log.tmp
+ls -l current.log.tmp | awk '{print $5}' > tmp
+read sz < tmp
+# Make a change
+${kadmin} -l mod -a requires-pre-auth recovtest@${R} || exit 1
+${kadmin} -l get recovtest@${R} | grep 'Attributes: requires-pre-auth$' > /dev/null || exit 1
+# Save the resulting log record
+ls -l current.log | awk '{print $5}' > tmp
+read nsz < tmp
+rm tmp
+dd bs=1 if=current.log skip=$sz of=current.log.tmp.saved-record count=`expr $nsz - $sz` 2>/dev/null
+# Undo the change
+${kadmin} -l mod -a -requires-pre-auth recovtest@${R} || exit 1
+${kadmin} -l get recovtest@${R} | grep 'Attributes:.$' > /dev/null || exit 1
+# Restore the log
+cp current.log current.log.save
+mv current.log.tmp current.log
+# Append the saved record
+cat current.log.tmp.saved-record >> current.log
+rm current.log.tmp.saved-record
+# Check that we still see the principal as modified
+${kadmin} -l get recovtest@${R} | grep 'Attributes: requires-pre-auth$' > /dev/null || exit 1
+
 # -- foo
 ipds=
 ipdm=
@@ -214,6 +243,12 @@ ${EGREP} 'up-to-date with version' iprop-slave-status >/dev/null || { echo "slav
 echo "checking for replay problems"
 ${EGREP} 'Entry already exists in database' messages.log && exit 1

+echo "compare versions on master and slave logs (no lock)"
+KRB5_CONFIG=${objdir}/krb5-slave.conf \
+${iprop_log} last-version -n > slave-last.tmp
+${iprop_log} last-version -n > master-last.tmp
+cmp master-last.tmp slave-last.tmp || exit 1
+
 echo "kill slave and remove log and database"
 sh ${leaks_kill} ipropd-slave $ipds || exit 1
 sleep 2
@@ -241,7 +276,7 @@ ${kadmin} -l cpw --random-password user@${R} > /dev/null || exit 1
 sleep 2

 echo "live truncate on master log"
-${iprop_log} truncate || exit 1
+${iprop_log} truncate -K 5 || exit 1
 sleep 2

 echo "Killing master and slave"
@@ -351,4 +386,4 @@ if [ "$db_type" = lmdb ]; then
    [ "`expr 1 + "$princs"`" -eq "$entries" ] || exit 1
 fi

-exit $ec
+exit 0
--- a/tests/kdc/krb5.conf.in
+++ b/tests/kdc/krb5.conf.in
@@ -97,6 +97,7 @@
 	signal_socket = @objdir@/signal
 	iprop-stats = @objdir@/iprop-stats
 	iprop-acl = @srcdir@/iprop-acl
+        log-max-size = 40000

 [hdb]
 	db-dir = @objdir@