def scan_prefix(prefix_obj, dns_servers=None, perform_dns=True, max_workers=100, ping_timeout=1, dns_settings=None, job_logger=None, skip_reserved=False):
    """
    Ping all existing IPs in a prefix. Creates/updates PingResult and SubnetScanResult.

    Pings run in a thread pool for speed; DB writes happen on the main thread
    to avoid exhausting PostgreSQL connections.

    Returns dict: {'total': int, 'up': int, 'down': int, 'skipped': int}
    """
    from .models import PingResult, PingHistory, SubnetScanResult, DnsHistory, ScanEvent

    log = job_logger or logger
    ip_addresses = list(prefix_obj.get_child_ips().select_related())

    # Partition: skip reserved IPs if setting is enabled
    if skip_reserved:
        pingable_ips = [ip for ip in ip_addresses if ip.status != 'reserved']
        skipped_ips = [ip for ip in ip_addresses if ip.status == 'reserved']
    else:
        pingable_ips = ip_addresses
        skipped_ips = []

    if skipped_ips:
        msg = f'Skipping {len(skipped_ips)} reserved IPs'
        log.info(msg)
        print(f'[Scan] {msg}', flush=True)

    # Phase 1: ping + DNS in parallel (no DB access)
    def _ping_ip(ip_obj):
        ip_str = str(ip_obj.address.ip)
        ping_data = ping_host(ip_str, timeout=ping_timeout)
        dns_name = ''
        dns_attempted = False
        if perform_dns and ping_data['is_reachable']:
            dns_attempted = True
            dns_name = resolve_dns(ip_str, dns_servers)
        return ip_obj, ping_data, dns_name, dns_attempted

    ping_results = []
    total_ips = len(pingable_ips)
    last_log_time = time.monotonic()
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(_ping_ip, ip_obj): ip_obj for ip_obj in pingable_ips}
        for future in concurrent.futures.as_completed(futures):
            try:
                ping_results.append(future.result())
            except Exception as e:
                log.error(f'Error pinging IP: {e}')
            now_mono = time.monotonic()
            if now_mono - last_log_time >= 60 or len(ping_results) == total_ips:
                up_so_far = sum(1 for _, r, _, _ in ping_results if r['is_reachable'])
                msg = f'Ping progress: {len(ping_results)}/{total_ips} ({up_so_far} up)'
                log.info(msg)
                print(f'[Scan] {msg}', flush=True)
                last_log_time = now_mono

    # Phase 2: bulk DB writes on the main thread
    now = timezone.now()

    # Pre-fetch all existing PingResults for all IPs (pingable + skipped)
    all_ip_ids = [ip_obj.pk for ip_obj in ip_addresses]
    existing_results = {}
    for i in range(0, len(all_ip_ids), 5000):
        batch = all_ip_ids[i:i + 5000]
        for pr in PingResult.objects.filter(ip_address_id__in=batch):
            existing_results[pr.ip_address_id] = pr

    to_update = []
    to_create = []
    history_records = []
    ips_to_dns_update = []
    dns_history_records = []
    scan_events = []

    for ip_obj, ping_data, dns_name, dns_attempted in ping_results:
        existing = existing_results.get(ip_obj.pk)

        if existing:
            # Capture state change before overwriting
            was_reachable = existing.is_reachable
            is_now_reachable = ping_data['is_reachable']
            if not existing.is_skipped:
                if was_reachable and not is_now_reachable:
                    scan_events.append(ScanEvent(
                        event_type='ip_went_down',
                        prefix=prefix_obj,
                        ip_address=ip_obj,
                        detail={
                            'dns_name': existing.dns_name or '',
                            'last_response_ms': existing.response_time_ms,
                        },
                    ))
                elif not was_reachable and is_now_reachable:
                    scan_events.append(ScanEvent(
                        event_type='ip_came_up',
                        prefix=prefix_obj,
                        ip_address=ip_obj,
                        detail={
                            'dns_name': dns_name or existing.dns_name or '',
                            'response_time_ms': ping_data['response_time_ms'],
                        },
                    ))

            existing.is_reachable = ping_data['is_reachable']
            existing.is_skipped = False
            existing.response_time_ms = ping_data['response_time_ms']
            existing.dns_name = dns_name if dns_name else (existing.dns_name or '')
            existing.last_checked = now
            if ping_data['is_reachable']:
                existing.last_seen = now
                existing.consecutive_down_count = 0
                existing.is_stale = False
            else:
                existing.consecutive_down_count += 1
            to_update.append(existing)
        else:
            to_create.append(PingResult(
                ip_address=ip_obj,
                is_reachable=ping_data['is_reachable'],
                is_skipped=False,
                response_time_ms=ping_data['response_time_ms'],
                dns_name=dns_name,
                last_checked=now,
                last_seen=now if ping_data['is_reachable'] else None,
                consecutive_down_count=0 if ping_data['is_reachable'] else 1,
            ))

        history_records.append(PingHistory(
            ip_address=ip_obj,
            is_reachable=ping_data['is_reachable'],
            response_time_ms=ping_data['response_time_ms'],
            dns_name=dns_name,
            checked_at=now,
        ))

        # DNS sync logic
        if dns_settings and dns_settings.dns_sync_to_netbox:
            current_netbox_dns = ip_obj.dns_name or ''
            should_update, new_value = _compute_dns_sync(
                dns_name, ping_data['is_reachable'], dns_attempted,
                current_netbox_dns, dns_settings,
            )
            if should_update:
                ip_obj.dns_name = new_value
                ips_to_dns_update.append(ip_obj)
                dns_history_records.append(DnsHistory(
                    ip_address=ip_obj,
                    old_dns_name=current_netbox_dns,
                    new_dns_name=new_value,
                    changed_at=now,
                ))
                scan_events.append(ScanEvent(
                    event_type='dns_changed',
                    prefix=prefix_obj,
                    ip_address=ip_obj,
                    detail={
                        'old_dns': current_netbox_dns,
                        'new_dns': new_value,
                    },
                ))

    # Phase 2b: create/update PingResult for skipped IPs (no history)
    for ip_obj in skipped_ips:
        existing = existing_results.get(ip_obj.pk)
        if existing:
            existing.is_reachable = False
            existing.is_skipped = True
            existing.response_time_ms = None
            existing.last_checked = now
            to_update.append(existing)
        else:
            to_create.append(PingResult(
                ip_address=ip_obj,
                is_reachable=False,
                is_skipped=True,
                response_time_ms=None,
                dns_name='',
                last_checked=now,
                last_seen=None,
            ))

    BATCH = 1000
    if to_update:
        update_fields = ['is_reachable', 'is_skipped', 'response_time_ms', 'dns_name', 'last_checked', 'last_seen', 'consecutive_down_count', 'is_stale']
        for i in range(0, len(to_update), BATCH):
            PingResult.objects.bulk_update(to_update[i:i + BATCH], update_fields)
        msg = f'Bulk updated {len(to_update)} existing results'
        log.info(msg)
        print(f'[Scan] {msg}', flush=True)

    if to_create:
        for i in range(0, len(to_create), BATCH):
            PingResult.objects.bulk_create(to_create[i:i + BATCH])
        msg = f'Bulk created {len(to_create)} new results'
        log.info(msg)
        print(f'[Scan] {msg}', flush=True)

    for i in range(0, len(history_records), BATCH):
        PingHistory.objects.bulk_create(history_records[i:i + BATCH])
    msg = f'Bulk created {len(history_records)} history records'
    log.info(msg)
    print(f'[Scan] {msg}', flush=True)

    # DNS sync: bulk update IPAddress.dns_name and create DnsHistory
    if ips_to_dns_update:
        from ipam.models import IPAddress as IPAddressModel
        for i in range(0, len(ips_to_dns_update), BATCH):
            IPAddressModel.objects.bulk_update(ips_to_dns_update[i:i + BATCH], ['dns_name'])
        msg = f'DNS sync: updated {len(ips_to_dns_update)} IPAddress dns_name fields'
        log.info(msg)
        print(f'[Scan] {msg}', flush=True)

    if dns_history_records:
        for i in range(0, len(dns_history_records), BATCH):
            DnsHistory.objects.bulk_create(dns_history_records[i:i + BATCH])
        msg = f'DNS sync: created {len(dns_history_records)} DNS history records'
        log.info(msg)
        print(f'[Scan] {msg}', flush=True)

    # Phase 3: create ScanEvent records for digest notifications
    if scan_events:
        for i in range(0, len(scan_events), BATCH):
            ScanEvent.objects.bulk_create(scan_events[i:i + BATCH])
        msg = f'Created {len(scan_events)} scan event(s) for digest'
        log.info(msg)
        print(f'[Scan] {msg}', flush=True)

    # Phase 4: Stale IP detection
    from .models import PrefixSchedule
    stale_settings = dns_settings  # reuse the PluginSettings object passed in
    stale_excluded = False
    if stale_settings:
        try:
            prefix_schedule = PrefixSchedule.objects.get(prefix=prefix_obj)
            if prefix_schedule.stale_mode == 'exclude':
                stale_excluded = True
        except PrefixSchedule.DoesNotExist:
            pass

    stale_events = []
    stale_to_update = []
    ips_to_remove = []

    if stale_settings and not stale_excluded:
        # Phase 4a: Tag stale IPs
        if stale_settings.stale_enabled:
            scans_thresh = stale_settings.stale_scans_threshold
            days_thresh = stale_settings.stale_days_threshold

            if scans_thresh > 0 or days_thresh > 0:
                # Get all down, non-stale PingResults for this prefix
                candidate_prs = PingResult.objects.filter(
                    ip_address__in=[ip.pk for ip in ip_addresses],
                    is_reachable=False,
                    is_skipped=False,
                    is_stale=False,
                ).select_related('ip_address')

                for pr in candidate_prs:
                    is_stale_now = False
                    if scans_thresh > 0 and pr.consecutive_down_count >= scans_thresh:
                        is_stale_now = True
                    if days_thresh > 0:
                        if pr.last_seen is not None:
                            if (now - pr.last_seen) >= timedelta(days=days_thresh):
                                is_stale_now = True
                        elif pr.last_checked is not None:
                            # Never seen online — use last_checked as reference
                            if (now - pr.last_checked) >= timedelta(days=days_thresh):
                                is_stale_now = True

                    if is_stale_now:
                        pr.is_stale = True
                        stale_to_update.append(pr)
                        stale_events.append(ScanEvent(
                            event_type='ip_went_stale',
                            prefix=prefix_obj,
                            ip_address=pr.ip_address,
                            detail={
                                'dns_name': pr.dns_name or '',
                                'consecutive_down_count': pr.consecutive_down_count,
                                'last_seen': str(pr.last_seen) if pr.last_seen else None,
                            },
                        ))

                if stale_to_update:
                    for i in range(0, len(stale_to_update), BATCH):
                        PingResult.objects.bulk_update(stale_to_update[i:i + BATCH], ['is_stale'])
                    msg = f'Tagged {len(stale_to_update)} IP(s) as stale'
                    log.info(msg)
                    print(f'[Scan] {msg}', flush=True)

        # Phase 4b: Auto-remove stale IPs
        if stale_settings.stale_remove_enabled and stale_settings.stale_remove_days > 0:
            remove_cutoff = now - timedelta(days=stale_settings.stale_remove_days)
            from django.db.models import Q
            removable_prs = PingResult.objects.filter(
                ip_address__in=[ip.pk for ip in ip_addresses],
                is_reachable=False,
                is_skipped=False,
                is_stale=True,
            ).filter(
                Q(last_seen__lte=remove_cutoff) | Q(last_seen__isnull=True)
            ).select_related('ip_address')

            for pr in removable_prs:
                stale_events.append(ScanEvent(
                    event_type='ip_removed_stale',
                    prefix=prefix_obj,
                    ip_address=pr.ip_address,
                    detail={
                        'dns_name': pr.dns_name or '',
                        'ip_address': str(pr.ip_address.address),
                        'last_seen': str(pr.last_seen) if pr.last_seen else None,
                    },
                ))
                ips_to_remove.append(pr.ip_address_id)

            if ips_to_remove:
                # Create events before deleting (ip_address FK will cascade)
                for i in range(0, len(stale_events), BATCH):
                    ScanEvent.objects.bulk_create(stale_events[i:i + BATCH])
                stale_events = []  # already saved
                from ipam.models import IPAddress as IPAddressModel
                deleted_count = IPAddressModel.objects.filter(pk__in=ips_to_remove).delete()[0]
                msg = f'Auto-removed {deleted_count} stale IP(s) from NetBox'
                log.info(msg)
                print(f'[Scan] {msg}', flush=True)

    # Save any remaining stale events
    if stale_events:
        for i in range(0, len(stale_events), BATCH):
            ScanEvent.objects.bulk_create(stale_events[i:i + BATCH])

    # Calculate actual subnet size for correct utilization
    network = ip_network(prefix_obj.prefix, strict=False)
    subnet_size = max(network.num_addresses - 2, 1) if network.prefixlen < 31 else network.num_addresses

    up = sum(1 for _, r, _, _ in ping_results if r['is_reachable'])
    skipped = len(skipped_ips)
    removed = len(ips_to_remove)
    tracked = len(ip_addresses)
    down = tracked - up - skipped - removed

    # Count stale IPs (remaining after removal)
    remaining_ip_pks = [ip.pk for ip in ip_addresses if ip.pk not in ips_to_remove] if ips_to_remove else [ip.pk for ip in ip_addresses]
    stale_count = PingResult.objects.filter(
        ip_address__in=remaining_ip_pks,
        is_stale=True,
    ).count()
    new_count = PingResult.objects.filter(
        ip_address__in=remaining_ip_pks,
        is_new=True,
    ).count()

    SubnetScanResult.objects.update_or_create(
        prefix=prefix_obj,
        defaults={
            'total_hosts': subnet_size,
            'hosts_up': up,
            'hosts_down': down,
            'hosts_skipped': skipped,
            'hosts_stale': stale_count,
            'hosts_new': new_count,
            'last_scanned': timezone.now(),
        },
    )

    state_changes = {}
    for event in scan_events:
        state_changes[event.event_type] = state_changes.get(event.event_type, 0) + 1
    if stale_to_update:
        state_changes['ip_went_stale'] = len(stale_to_update)
    if ips_to_remove:
        state_changes['ip_removed_stale'] = len(ips_to_remove)

    return {'total': tracked, 'up': up, 'down': down, 'skipped': skipped, 'stale': stale_count, 'removed': removed, 'state_changes': state_changes}