do not accept new records that have older timestamps than ones in dat…

…abase

when adding the records.

kcndbd/kcndb_db.c

@@ -347,10 +347,31 @@ kcndb_db_record_read(struct kcndb_db_table *kdt, struct kcndb_db_record *kdr)
 	return true;
 }
 
+static bool
+kcndb_db_record_read_last(struct kcndb_db_table *kdt,
+    struct kcndb_db_record *kdr)
+{
+	struct kcndb_db_base *kdb;
+	size_t off;
+
+	kdb = &kcndb_db_base[TYPE2INDEX(kdt->kdt_type)];
+	if (kdb->kdb_tablesize < KCNDB_DB_RECORDSIZ) {
+		kdr->kdr_time = 0;
+		kdr->kdr_val = 0;
+		kdr->kdr_locidx = 0;
+		return true;
+	}
+	off = kdb->kdb_tablesize - KCNDB_DB_RECORDSIZ;
+	if (! kcndb_file_seek_head(kdt->kdt_table, off))
+		return false;
+	return kcndb_db_record_read(kdt, kdr);
+}
+
 bool
 kcndb_db_record_add(struct kcndb_db *kd, enum kcn_eq_type type,
     struct kcndb_db_record *kdr)
 {
+	struct kcndb_db_record kdr0;
 	struct kcndb_db_table *kdt;
 	struct kcn_buf *kb;
 	bool rc;
@@ -359,6 +380,14 @@ kcndb_db_record_add(struct kcndb_db *kd, enum kcn_eq_type type,
 	kb = kcndb_file_buf(kdt->kdt_table);
 	if (! kcndb_db_wrlock(kdt))
 		return false;
+	rc = kcndb_db_record_read_last(kdt, &kdr0);
+	if (! rc)
+		goto out;
+	if (kdr0.kdr_time > kdr->kdr_time) {
+		errno = ERANGE; /* XXX */
+		rc = false;
+		goto out;
+	}
 	rc = kcndb_db_loc_add(kdt, kdr->kdr_loc, kdr->kdr_loclen,
 	    &kdr->kdr_locidx);
 	if (! rc)