diff --git a/src/dmtest/fs.py b/src/dmtest/fs.py
index 8a3f6e1..dac7f01 100644
--- a/src/dmtest/fs.py
+++ b/src/dmtest/fs.py
@@ -61,7 +61,8 @@ def check_cmd(self):
 
     def mkfs_cmd(self, opts):
         discard_arg = "discard" if opts.get("discard", True) else "nodiscard"
-        return f"mkfs.ext4 -F -E lazy_itable_init=1,{discard_arg} {self._dev}"
+        lazy_init = 1 if opts.get("lazy_itable_init", True) else 0
+        return f"mkfs.ext4 -F -E lazy_itable_init={lazy_init},{discard_arg} {self._dev}"
 
 
 class Xfs(BaseFS):
diff --git a/src/dmtest/vdo/basic_01_tests.py b/src/dmtest/vdo/basic_01_tests.py
new file mode 100644
index 0000000..e74aa19
--- /dev/null
+++ b/src/dmtest/vdo/basic_01_tests.py
@@ -0,0 +1,60 @@
+"""VDO basic functional test.
+
+Verifies VDO persistence by writing data to a filesystem on VDO, stopping
+the VDO device, restarting it, and verifying the data is still readable.
+"""
+from dmtest.assertions import assert_equal, assert_string_in
+from dmtest.utils import get_dmesg_log
+from dmtest.vdo.utils import standard_vdo, standard_stack, mounted_fs
+import dmtest.process as process
+import time
+
+def t_basic(fix):
+    """Basic VDO functional test: write files, stop/start VDO, verify data persists."""
+    # Create VDO with slab_bits=17 (SLAB_BITS_SMALL)
+    with standard_vdo(fix, slab_bits=17) as vdo:
+        with mounted_fs(vdo.path, format=True) as mount_point:
+            # Create file foo1 with "Hello World"
+            file1 = f"{mount_point}/foo1"
+            process.run(f"bash -c 'echo Hello World > {file1}'")
+
+            # Create subdirectory dir2
+            dir2 = f"{mount_point}/dir2"
+            process.run(f"mkdir {dir2}")
+
+            # Copy foo1 to dir2/foo2
+            file2 = f"{dir2}/foo2"
+            process.run(f"cp {file1} {file2}")
+
+            # Copy foo1 to foo3
+            file3 = f"{mount_point}/foo3"
+            process.run(f"cp {file1} {file3}")
+
+            # Drop caches
+            process.run("echo 1 > /proc/sys/vm/drop_caches")
+
+            # Verify content of foo1 and foo2
+            result1 = process.run(f"cat {file1}")
+            assert_equal(result1[1].strip(), "Hello World")
+
+            result2 = process.run(f"cat {file2}")
+            assert_equal(result2[1].strip(), "Hello World")
+
+    # VDO device is now stopped (exited context manager)
+    # Get kernel log timestamp before restarting
+    start_time = time.time()
+
+    # Restart VDO device without reformatting
+    with standard_vdo(fix, format=False, slab_bits=17) as vdo:
+        with mounted_fs(vdo.path) as mount_point:
+            # Verify content of foo3
+            file3 = f"{mount_point}/foo3"
+            result3 = process.run(f"cat {file3}")
+            assert_equal(result3[1].strip(), "Hello World")
+
+    # Check kernel log for VDO startup message
+    log_message = get_dmesg_log(start_time)
+    assert_string_in(log_message, "VDO commencing normal operation")
+
+def register(tests):
+    tests.register("/vdo/basic/basic01", t_basic)
diff --git a/src/dmtest/vdo/basic_fs_dedupe_tests.py b/src/dmtest/vdo/basic_fs_dedupe_tests.py
new file mode 100644
index 0000000..1c7f989
--- /dev/null
+++ b/src/dmtest/vdo/basic_fs_dedupe_tests.py
@@ -0,0 +1,88 @@
+"""
+VDO BasicFSDedupe test - Filesystem-level deduplication verification
+"""
+import logging as log
+import os
+import shutil
+import tempfile
+
+from dmtest.assertions import assert_near
+from dmtest.gendatablocks import make_block_range
+from dmtest.vdo.stats import vdo_stats, make_delta_stats
+from dmtest.vdo.utils import standard_vdo, fsync, mounted_fs
+
+
+def t_basic_fs_dedupe(fix) -> None:
+    """
+    Basic filesystem-level deduplication test that writes a dataset twice
+    and verifies deduplication achieves expected space savings.
+    """
+    MB = 1024 * 1024
+    num_files = 32
+    file_size_mb = 8
+    blocks_per_file = file_size_mb * MB // 4096  # 8MB / 4KB = 2048 blocks
+
+    with standard_vdo(fix) as vdo:
+        with mounted_fs(vdo.path, format=True, lazy_itable_init=False) as mount_point:
+            # Create subdirectories on VDO filesystem
+            original_dir = os.path.join(mount_point, "original")
+            copy1_dir = os.path.join(mount_point, "copy1")
+            os.makedirs(original_dir)
+            os.makedirs(copy1_dir)
+
+            # Record initial stats after filesystem setup
+            fsync(vdo.path)
+            initial_stats = vdo_stats(vdo)
+
+            # Generate dataset in a scratch directory
+            with tempfile.TemporaryDirectory() as scratch_dir:
+                dataset_dir = os.path.join(scratch_dir, "dataset")
+                os.makedirs(dataset_dir)
+
+                log.info(f"Generating dataset: {num_files} files × {file_size_mb}MB each = 256MB total")
+                for i in range(num_files):
+                    file_path = os.path.join(dataset_dir, f"file_{i:08d}")
+                    # Create the file first
+                    with open(file_path, 'w') as f:
+                        pass
+                    # Write data to the file
+                    block_range = make_block_range(file_path, blocks_per_file)
+                    block_range.write(f"BFD{i:04d}", dedupe=0.0, fsync=False)
+
+                # Copy dataset to "original" directory
+                log.info("Copying dataset to 'original' directory")
+                shutil.copytree(dataset_dir, os.path.join(original_dir, "data"))
+
+                # Sync and check stats after first write
+                fsync(vdo.path)
+                stats_after_first = vdo_stats(vdo)
+                delta_first = make_delta_stats(stats_after_first, initial_stats)
+
+                data_blocks = delta_first['dataBlocksUsed']
+                logical_blocks = delta_first['logicalBlocksUsed']
+                ratio_first = data_blocks / logical_blocks if logical_blocks > 0 else 0
+
+                log.info(f"After first write: data={data_blocks}, logical={logical_blocks}, ratio={ratio_first:.3f}")
+                # Verify minimal deduplication on first write (filesystem metadata may cause some variance)
+                assert_near(ratio_first, 1.0, 0.1, "Data-to-logical ratio after first write")
+
+                # Copy the same dataset to "copy1" directory (duplicate copy)
+                log.info("Copying dataset to 'copy1' directory (duplicate)")
+                shutil.copytree(dataset_dir, os.path.join(copy1_dir, "data"))
+
+                # Sync and check stats after second write
+                fsync(vdo.path)
+                stats_after_second = vdo_stats(vdo)
+                delta_second = make_delta_stats(stats_after_second, initial_stats)
+
+                data_blocks_2 = delta_second['dataBlocksUsed']
+                logical_blocks_2 = delta_second['logicalBlocksUsed']
+                ratio_second = data_blocks_2 / logical_blocks_2 if logical_blocks_2 > 0 else 0
+
+                log.info(f"After second write: data={data_blocks_2}, logical={logical_blocks_2}, ratio={ratio_second:.3f}")
+                # Verify significant deduplication on second write (~50% ratio expected)
+                assert_near(ratio_second, 0.5, 0.05, "Data-to-logical ratio after second write (with dedupe)")
+
+
+def register(tests):
+    tests.register("/vdo/basic/fs-dedupe", t_basic_fs_dedupe)
diff --git a/src/dmtest/vdo/create_03_tests.py b/src/dmtest/vdo/create_03_tests.py
new file mode 100644
index 0000000..c002a8f
--- /dev/null
+++ b/src/dmtest/vdo/create_03_tests.py
@@ -0,0 +1,70 @@
+"""
+VDO Create03 test - Device lifecycle stability and logging verification
+"""
+import logging as log
+import re
+import time
+
+from dmtest.assertions import assert_string_in
+from dmtest.utils import get_dmesg_log
+from dmtest.vdo.utils import standard_stack
+import dmtest.process as process
+
+
+def t_create_03(fix) -> None:
+    """
+    Test creating and tearing down VDO devices many times to verify device
+    lifecycle stability and proper logging behavior. Also verifies that VDO
+    messages include device instance numbers in kernel logs.
+    """
+    iteration_count = 1024
+
+    log.info(f"Starting Create03 test with {iteration_count} iterations")
+
+    # Create the VDO stack (formats the device)
+    stack = standard_stack(fix, slab_bits=17)
+
+    for i in range(iteration_count):
+        # Record time before starting device
+        start_time = time.time()
+
+        # Activate the VDO device
+        vdo = stack.activate()
+
+        # After first iteration, check kernel logs
+        if i > 0:
+            log.info(f"Iteration {i + 1}/{iteration_count}: Checking kernel logs")
+            kernel_log = get_dmesg_log(start_time)
+
+            # Verify that VDO messages are present (pattern: "vdo[0-9]+:")
+            # This regex looks for lines like "vdo0: ..." or "vdo1: ..."
+            vdo_messages = [line for line in kernel_log.split('\n')
+                           if re.search(r'vdo[0-9]+:', line)]
+
+            if vdo_messages:
+                # Verify that VDO messages include device instance number
+                # Pattern: "vdo([0-9]+:|:\[SI\]:)"
+                # This matches "vdo0:", "vdo1:", "vdo:S:", "vdo:I:", etc.
+                for msg in vdo_messages:
+                    # Messages from interrupt context may be anonymous (vdo:S: or vdo:I:)
+                    # so we allow those, but most messages should have instance numbers
+                    if not re.search(r'vdo([0-9]+:|:\[[SI]\]:)', msg):
+                        log.warning(f"VDO message without instance number: {msg}")
+        elif i == 0:
+            log.info(f"Iteration {i + 1}/{iteration_count}: First iteration (no log check)")
+
+        # Log progress periodically
+        if (i + 1) % 100 == 0:
+            log.info(f"Completed {i + 1}/{iteration_count} iterations")
+
+        # Stop the VDO device
+        vdo.remove()
+
+        # For the next iteration, we don't need to format (already formatted)
+        stack = standard_stack(fix, format=False, slab_bits=17)
+
+    log.info(f"Create03 test completed successfully after {iteration_count} iterations")
+
+
+def register(tests):
+    tests.register("/vdo/creation/create-03", t_create_03)
diff --git a/src/dmtest/vdo/register.py b/src/dmtest/vdo/register.py
index 0df5542..b415deb 100644
--- a/src/dmtest/vdo/register.py
+++ b/src/dmtest/vdo/register.py
@@ -1,14 +1,22 @@
+import dmtest.vdo.basic_01_tests as vdo_basic_01
+import dmtest.vdo.basic_fs_dedupe_tests as vdo_basic_fs_dedupe
 import dmtest.vdo.compress_tests as vdo_compress
+import dmtest.vdo.create_03_tests as vdo_create_03
 import dmtest.vdo.creation_tests as vdo_creation
 import dmtest.vdo.dedupe_tests as vdo_dedupe
 import dmtest.vdo.full_tests as vdo_full
 import dmtest.vdo.load_failure_tests as vdo_load_failure
 import dmtest.vdo.recovery_tests as vdo_recovery
+import dmtest.vdo.uds_timeout_tests as vdo_uds_timeout
 
 def register(tests):
+    vdo_basic_01.register(tests)
+    vdo_basic_fs_dedupe.register(tests)
+    vdo_create_03.register(tests)
     vdo_creation.register(tests)
     vdo_dedupe.register(tests)
     vdo_compress.register(tests)
     vdo_full.register(tests)
     vdo_load_failure.register(tests)
     vdo_recovery.register(tests)
+    vdo_uds_timeout.register(tests)
diff --git a/src/dmtest/vdo/uds_timeout_tests.py b/src/dmtest/vdo/uds_timeout_tests.py
new file mode 100644
index 0000000..5283157
--- /dev/null
+++ b/src/dmtest/vdo/uds_timeout_tests.py
@@ -0,0 +1,148 @@
+"""VDO UDS deduplication timeout test.
+
+Tests that VDO reports dedupe advice timeouts when the UDS index
+cannot respond quickly enough due to slow underlying storage.
+Converted from UDSTimeout01.pm.
+"""
+import logging as log
+import os
+import threading
+import time
+
+import dmtest.device_mapper.dev as dmdev
+import dmtest.device_mapper.table as table
+import dmtest.device_mapper.targets as targets
+import dmtest.process as process
+import dmtest.vdo.stats as vdo_stats_mod
+import dmtest.vdo.vdo_stack as vs
+from dmtest.gendatablocks import make_block_range
+from dmtest.utils import dev_size
+from dmtest.vdo.utils import wait_for_index, BLOCK_SIZE
+
+BLOCK_COUNT = 20000
+DATASET_COUNT = 10
+READ_DELAY_MS = 6000
+DELAY_ACTIVE_SECS = 60
+
+
+def _swap_delay_table(delay_dev_name, data_size, data_dev, read_delay_ms):
+    """Swap the dm-delay table to change read delay without udev blocking."""
+    tline = f"0 {data_size} delay {data_dev} 0 {read_delay_ms} {data_dev} 0 0"
+    process.run(f"dmsetup suspend --noflush {delay_dev_name}")
+    process.run(f"dmsetup load {delay_dev_name} --table '{tline}'")
+    process.run(f"dmsetup resume --noudevsync {delay_dev_name}")
+
+
+def t_uds_timeout(fix) -> None:
+    """Test that VDO reports dedupe timeouts with slow storage.
+
+    Writes duplicate data to a VDO stacked on a dm-delay device and
+    verifies that dedupe advice timeouts increase when UDS index reads
+    are delayed beyond the 5-second default timeout.
+    """
+    data_dev = fix.cfg["data_dev"]
+    data_size = dev_size(data_dev)
+
+    # Phase 1: Format VDO directly and populate the UDS index.
+    log.info("Phase 1: populating UDS index on fast storage")
+    stack = vs.VDOStack(data_dev)
+    with stack.activate() as vdo:
+        wait_for_index(vdo)
+        vdo_path = str(vdo)
+
+        log.info(f"Writing {DATASET_COUNT} datasets of {BLOCK_COUNT} blocks each")
+        for n in range(DATASET_COUNT):
+            tag = f"D{n}"
+            first_offset = 2 * n * BLOCK_COUNT
+            log.info(f"Writing dataset {tag} at offset {first_offset}")
+            br = make_block_range(vdo_path, BLOCK_COUNT, BLOCK_SIZE, first_offset)
+            br.write(tag=tag, fsync=True)
+
+        before_stats = vdo_stats_mod.vdo_stats(vdo)
+        before_timeouts = before_stats['dedupeAdviceTimeouts']
+        log.info(f"Dedupe advice timeouts after phase 1: {before_timeouts}")
+
+    # Phase 2: Restart VDO on a dm-delay device with read delay.
+    # Create dm-delay initially with 0ms delay to avoid udev probe hang,
+    # then swap to the real delay after VDO is running and caches are dropped.
+    log.info("Phase 2: restarting VDO on dm-delay device")
+    zero_delay_table = table.Table(
+        targets.Target("delay", data_size,
+                       data_dev, 0, 0, data_dev, 0, 0)
+    )
+    delay_dev = dmdev.dev(zero_delay_table)
+    try:
+        stack2 = vs.VDOStack(str(delay_dev), format=False)
+        with stack2.activate() as vdo:
+            log.info("Waiting for UDS index to come online")
+            wait_for_index(vdo)
+            vdo_path = str(vdo)
+
+            # Evict UDS index pages from the page cache so subsequent
+            # lookups must read through the slow dm-delay device.
+            log.info("Dropping page caches")
+            with open("/proc/sys/vm/drop_caches", "w") as f:
+                f.write("3\n")
+
+            # Enable read delay on the underlying device.
+            log.info(f"Enabling {READ_DELAY_MS}ms read delay")
+            _swap_delay_table(delay_dev.name, data_size, data_dev,
+                              READ_DELAY_MS)
+
+            # Write duplicate copies of all datasets in parallel.
+            log.info("Writing second copies of all datasets in parallel")
+            errors = []
+
+            def write_second_copy(dataset_num: int) -> None:
+                try:
+                    tag = f"D{dataset_num}"
+                    second_offset = 2 * dataset_num * BLOCK_COUNT + BLOCK_COUNT
+                    br = make_block_range(
+                        vdo_path, BLOCK_COUNT, BLOCK_SIZE, second_offset
+                    )
+                    br.write(tag=tag)
+                except Exception as e:
+                    errors.append(e)
+
+            threads = []
+            for n in range(DATASET_COUNT):
+                t = threading.Thread(target=write_second_copy, args=(n,))
+                threads.append(t)
+                t.start()
+
+            # Keep the delay active long enough for VDO to process
+            # blocks through the slow path and accumulate timeouts,
+            # then disable it so remaining I/O drains quickly.
+            log.info(f"Waiting {DELAY_ACTIVE_SECS}s for timeouts to accumulate")
+            time.sleep(DELAY_ACTIVE_SECS)
+
+            log.info("Disabling read delay for remaining I/O and cleanup")
+            _swap_delay_table(delay_dev.name, data_size, data_dev, 0)
+
+            for t in threads:
+                t.join()
+
+            if errors:
+                raise errors[0]
+
+            os.sync()
+
+            after_stats = vdo_stats_mod.vdo_stats(vdo)
+            after_timeouts = after_stats['dedupeAdviceTimeouts']
+            log.info(f"Dedupe advice timeouts after phase 2: {after_timeouts}")
+
+            assert after_timeouts > before_timeouts, (
+                f"Expected dedupe advice timeouts to increase, "
+                f"but before={before_timeouts}, after={after_timeouts}"
+            )
+            log.info(
+                f"Timeout count increased from {before_timeouts} to {after_timeouts}"
+            )
+    finally:
+        delay_dev.remove()
+
+
+def register(tests):
+    tests.register_batch("/vdo/uds-timeout/", [
+        ("timeout", t_uds_timeout),
+    ])
diff --git a/src/dmtest/vdo/utils.py b/src/dmtest/vdo/utils.py
index dce3dca..ed25144 100644
--- a/src/dmtest/vdo/utils.py
+++ b/src/dmtest/vdo/utils.py
@@ -3,8 +3,10 @@
 import dmtest.vdo.vdo_stack as vs
 import dmtest.vdo.stats as stats
 import dmtest.vdo.status as status
+from dmtest.fs import Ext4
 
 import code
+from contextlib import contextmanager
 import json
 import logging as log
 from math import ceil
@@ -19,6 +21,9 @@
 
 BLOCK_SIZE = 4 * kB
 
+# VDO slab bit count constants
+SLAB_BITS_SMALL = 17  # Smallest size that works for any RSVP-reserved host
+
 fio_config_template = """
 [stuff]
 randrepeat=1
@@ -53,11 +58,35 @@ def wait_for_index(dev):
     if status.vdo_status(dev)["index-state"] != "online":
         raise AssertionError("VDO not online within 30 seconds")
 
+
+
+
+
+
+@contextmanager
+def mounted_fs(dev, fs_class=None, format=False, **format_opts):
+
+    Yields the mount point path and ensures unmount, fsck, and mount
+    point removal on exit.
+    if fs_class is None:
+        fs_class = Ext4
+    fs = fs_class(dev)
+    if format:
+        fs.format(**format_opts)
+    with tempfile.TemporaryDirectory() as mount_point:
+        fs.mount(mount_point)
+        try:
+            yield mount_point
+        finally:
+            fs.umount()
+
+
 def fsync(dev):
     """Sync the specified device or file."""
     with open(dev, 'w') as thing:
         os.fsync(thing.fileno())
 
+
 def run_fio_with_config(fio_config, raise_on_fail=True):
     """Run fio with the specified config file content.
 
diff --git a/test_dependencies.toml b/test_dependencies.toml
index 2a4679c..0c7e8a5 100644
--- a/test_dependencies.toml
+++ b/test_dependencies.toml
@@ -925,6 +925,40 @@ targets = [
     "thin-pool",
 ]
 
+["/vdo/basic/basic01"]
+executables = [
+    "bash",
+    "blockdev",
+    "cat",
+    "cp",
+    "dmsetup",
+    "echo",
+    "fsck.ext4",
+    "mkdir",
+    "mkfs.ext4",
+    "mount",
+    "umount",
+    "vdoformat",
+]
+targets = [
+    "vdo",
+]
+
+["/vdo/basic/fs-dedupe"]
+executables = [
+    "blockdev",
+    "dmsetup",
+    "echo",
+    "fsck.ext4",
+    "mkfs.ext4",
+    "mount",
+    "umount",
+    "vdoformat",
+]
+targets = [
+    "vdo",
+]
+
 ["/vdo/compress/compress"]
 executables = [
     "blkdiscard",
@@ -936,6 +970,16 @@ targets = [
     "vdo",
 ]
 
+["/vdo/creation/create-03"]
+executables = [
+    "blockdev",
+    "dmsetup",
+    "vdoformat",
+]
+targets = [
+    "vdo",
+]
+
 ["/vdo/creation/create01"]
 executables = [
     "blockdev",
@@ -1065,3 +1109,14 @@ targets = [
     "linear",
     "vdo",
 ]
+
+["/vdo/uds-timeout/timeout"]
+executables = [
+    "blockdev",
+    "dmsetup",
+    "vdoformat",
+]
+targets = [
+    "delay",
+    "vdo",
+]